Mirrors/servo
1
0
Fork 0
mirror of https://github.com/servo/servo.git synced 2025-07-22 23:03:42 +01:00
Code Issues Projects Releases Packages Wiki Activity

Properly implement TaskSource for NetworkingTaskSource

Browse source
This commit is contained in:
Keith Yeung 2016-10-13 17:41:48 -07:00
parent d99d26cf1f
commit 72cb856e31
15 changed files with 89 additions and 69 deletions
Download patch file Download diff file Expand all files Collapse all files

4
components/script/dom/bluetooth.rs
View file

@ -148,14 +148,14 @@ pub fn response_async<T: AsyncBluetoothListener + Reflectable + 'static>(
promise: &Rc<Promise>,
receiver: &T) -> IpcSender<BluetoothResponseResult> {
let (action_sender, action_receiver) = ipc::channel().unwrap();
let chan = receiver.global().networking_task_source();
let task_source = receiver.global().networking_task_source();
let context = Arc::new(Mutex::new(BluetoothContext {
promise: Some(TrustedPromise::new(promise.clone())),
receiver: Trusted::new(receiver),
}));
let listener = NetworkListener {
context: context,
script_chan: chan,
task_source: task_source,
wrapper: None,
};
ROUTER.add_route(action_receiver.to_opaque(), box move |message| {

4
components/script/dom/eventsource.rs
View file

@ -132,8 +132,8 @@ impl EventSource {
let context = EventSourceContext;
let listener = NetworkListener {
context: Arc::new(Mutex::new(context)),
script_chan: global.script_chan(),
wrapper: None
task_source: global.networking_task_source(),
wrapper: Some(global.get_runnable_wrapper())
};
let (action_sender, action_receiver) = ipc::channel().unwrap();
ROUTER.add_route(action_receiver.to_opaque(), box move |message| {

5
components/script/dom/globalscope.rs
View file

@ -42,6 +42,7 @@ use std::collections::hash_map::Entry;
use std::ffi::CString;
use std::panic;
use task_source::file_reading::FileReadingTaskSource;
use task_source::networking::NetworkingTaskSource;
use time::{Timespec, get_time};
use timers::{IsInterval, OneshotTimerCallback, OneshotTimerHandle};
use timers::{OneshotTimers, TimerCallback};
@ -325,12 +326,12 @@ impl GlobalScope {
/// `ScriptChan` to send messages to the networking task source of
/// this of this global scope.
pub fn networking_task_source(&self) -> Box<ScriptChan + Send> {
pub fn networking_task_source(&self) -> NetworkingTaskSource {
if let Some(window) = self.downcast::<Window>() {
return window.networking_task_source();
}
if let Some(worker) = self.downcast::<WorkerGlobalScope>() {
return worker.script_chan();
return worker.networking_task_source();
}
unreachable!();
}

8
components/script/dom/htmlimageelement.rs
View file

@ -26,8 +26,6 @@ use ipc_channel::ipc;
use ipc_channel::router::ROUTER;
use net_traits::image::base::{Image, ImageMetadata};
use net_traits::image_cache_thread::{ImageResponder, ImageResponse};
use script_runtime::CommonScriptMsg;
use script_runtime::ScriptThreadEventCategory::UpdateReplacedElement;
use script_thread::Runnable;
use std::i32;
use std::sync::Arc;
@ -140,7 +138,7 @@ impl HTMLImageElement {
let trusted_node = Trusted::new(self);
let (responder_sender, responder_receiver) = ipc::channel().unwrap();
let script_chan = window.networking_task_source();
let task_source = window.networking_task_source();
let wrapper = window.get_runnable_wrapper();
ROUTER.add_route(responder_receiver.to_opaque(), box move |message| {
// Return the image via a message to the script thread, which marks the element
@ -148,9 +146,7 @@ impl HTMLImageElement {
let image_response = message.to().unwrap();
let runnable = box ImageResponseHandlerRunnable::new(
trusted_node.clone(), image_response);
let runnable = wrapper.wrap_runnable(runnable);
let _ = script_chan.send(CommonScriptMsg::RunnableMsg(
UpdateReplacedElement, runnable));
let _ = task_source.queue_with_wrapper(runnable, &wrapper);
});
image_cache.request_image_and_metadata(img_url,

4
components/script/dom/htmllinkelement.rs
View file

@ -243,8 +243,8 @@ impl HTMLLinkElement {
let (action_sender, action_receiver) = ipc::channel().unwrap();
let listener = NetworkListener {
context: context,
script_chan: document.window().networking_task_source(),
wrapper: Some(document.window().get_runnable_wrapper()),
task_source: document.window().networking_task_source(),
wrapper: Some(document.window().get_runnable_wrapper())
};
ROUTER.add_route(action_receiver.to_opaque(), box move |message| {
listener.notify_fetch(message.to().unwrap());

5
components/script/dom/htmlmediaelement.rs
View file

@ -521,11 +521,10 @@ impl HTMLMediaElement {
let context = Arc::new(Mutex::new(HTMLMediaElementContext::new(self, url.clone())));
let (action_sender, action_receiver) = ipc::channel().unwrap();
let window = window_from_node(self);
let script_chan = window.networking_task_source();
let listener = NetworkListener {
context: context,
script_chan: script_chan,
wrapper: Some(window.get_runnable_wrapper()),
task_source: window.networking_task_source(),
wrapper: Some(window.get_runnable_wrapper())
};
ROUTER.add_route(action_receiver.to_opaque(), box move |message| {

4
components/script/dom/htmlscriptelement.rs
View file

@ -262,8 +262,8 @@ fn fetch_a_classic_script(script: &HTMLScriptElement,
let (action_sender, action_receiver) = ipc::channel().unwrap();
let listener = NetworkListener {
context: context,
script_chan: doc.window().networking_task_source(),
wrapper: Some(doc.window().get_runnable_wrapper()),
task_source: doc.window().networking_task_source(),
wrapper: Some(doc.window().get_runnable_wrapper())
};
ROUTER.add_route(action_receiver.to_opaque(), box move |message| {

38
components/script/dom/websocket.rs
View file

@ -33,14 +33,16 @@ use net_traits::CoreResourceMsg::{SetCookiesForUrl, WebsocketConnect};
use net_traits::MessageData;
use net_traits::hosts::replace_hosts;
use net_traits::unwrap_websocket_protocol;
use script_runtime::{CommonScriptMsg, ScriptChan};
use script_runtime::CommonScriptMsg;
use script_runtime::ScriptThreadEventCategory::WebSocketEvent;
use script_thread::Runnable;
use script_thread::{Runnable, RunnableWrapper};
use std::ascii::AsciiExt;
use std::borrow::ToOwned;
use std::cell::Cell;
use std::ptr;
use std::thread;
use task_source::TaskSource;
use task_source::networking::NetworkingTaskSource;
use websocket::client::request::Url;
use websocket::header::{Headers, WebSocketProtocol};
use websocket::ws::util::url::parse_url;
@ -141,7 +143,8 @@ mod close_code {
}
pub fn close_the_websocket_connection(address: Trusted<WebSocket>,
sender: Box<ScriptChan>,
task_source: &NetworkingTaskSource,
wrapper: &RunnableWrapper,
code: Option<u16>,
reason: String) {
let close_task = box CloseTask {
@ -150,17 +153,19 @@ pub fn close_the_websocket_connection(address: Trusted<WebSocket>,
code: code,
reason: Some(reason),
};
sender.send(CommonScriptMsg::RunnableMsg(WebSocketEvent, close_task)).unwrap();
task_source.queue_with_wrapper(close_task, &wrapper).unwrap();
}
pub fn fail_the_websocket_connection(address: Trusted<WebSocket>, sender: Box<ScriptChan>) {
pub fn fail_the_websocket_connection(address: Trusted<WebSocket>,
task_source: &NetworkingTaskSource,
wrapper: &RunnableWrapper) {
let close_task = box CloseTask {
address: address,
failed: true,
code: Some(close_code::ABNORMAL),
reason: None,
};
sender.send(CommonScriptMsg::RunnableMsg(WebSocketEvent, close_task)).unwrap();
task_source.queue_with_wrapper(close_task, &wrapper).unwrap();
}
#[dom_struct]
@ -268,7 +273,8 @@ impl WebSocket {
*ws.sender.borrow_mut() = Some(dom_action_sender);
let moved_address = address.clone();
let sender = global.networking_task_source();
let task_source = global.networking_task_source();
let wrapper = global.get_runnable_wrapper();
thread::spawn(move || {
while let Ok(event) = dom_event_receiver.recv() {
match event {
@ -278,20 +284,22 @@ impl WebSocket {
headers: headers,
protocols: protocols,
};
sender.send(CommonScriptMsg::RunnableMsg(WebSocketEvent, open_thread)).unwrap();
task_source.queue_with_wrapper(open_thread, &wrapper).unwrap();
},
WebSocketNetworkEvent::MessageReceived(message) => {
let message_thread = box MessageReceivedTask {
address: moved_address.clone(),
message: message,
};
sender.send(CommonScriptMsg::RunnableMsg(WebSocketEvent, message_thread)).unwrap();
task_source.queue_with_wrapper(message_thread, &wrapper).unwrap();
},
WebSocketNetworkEvent::Fail => {
fail_the_websocket_connection(moved_address.clone(), sender.clone());
fail_the_websocket_connection(moved_address.clone(),
&task_source, &wrapper);
},
WebSocketNetworkEvent::Close(code, reason) => {
close_the_websocket_connection(moved_address.clone(), sender.clone(), code, reason);
close_the_websocket_connection(moved_address.clone(),
&task_source, &wrapper, code, reason);
},
}
}
@ -436,8 +444,8 @@ impl WebSocketMethods for WebSocket {
self.ready_state.set(WebSocketRequestState::Closing);
let address = Trusted::new(self);
let sender = self.global().networking_task_source();
fail_the_websocket_connection(address, sender);
let task_source = self.global().networking_task_source();
fail_the_websocket_connection(address, &task_source, &self.global().get_runnable_wrapper());
}
WebSocketRequestState::Open => {
self.ready_state.set(WebSocketRequestState::Closing);
@ -470,8 +478,8 @@ impl Runnable for ConnectionEstablishedTask {
// Step 1: Protocols.
if !self.protocols.is_empty() && self.headers.get::<WebSocketProtocol>().is_none() {
let sender = ws.global().networking_task_source();
fail_the_websocket_connection(self.address, sender);
let task_source = ws.global().networking_task_source();
fail_the_websocket_connection(self.address, &task_source, &ws.global().get_runnable_wrapper());
return;
}

2
components/script/dom/window.rs
View file

@ -267,7 +267,7 @@ impl Window {
self.user_interaction_task_source.clone()
}
pub fn networking_task_source(&self) -> Box<ScriptChan + Send> {
pub fn networking_task_source(&self) -> NetworkingTaskSource {
self.networking_task_source.clone()
}

5
components/script/dom/workerglobalscope.rs
View file

@ -41,6 +41,7 @@ use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::Receiver;
use task_source::file_reading::FileReadingTaskSource;
use task_source::networking::NetworkingTaskSource;
use timers::{IsInterval, TimerCallback};
use url::Url;
@ -361,6 +362,10 @@ impl WorkerGlobalScope {
FileReadingTaskSource(self.script_chan())
}
pub fn networking_task_source(&self) -> NetworkingTaskSource {
NetworkingTaskSource(self.script_chan())
}
pub fn new_script_pair(&self) -> (Box<ScriptChan + Send>, Box<ScriptPort + Send>) {
let dedicated = self.downcast::<DedicatedWorkerGlobalScope>();
if let Some(dedicated) = dedicated {

23
components/script/dom/xmlhttprequest.rs
View file

@ -49,13 +49,12 @@ use js::jsapi::{JSContext, JS_ParseJSON};
use js::jsapi::JS_ClearPendingException;
use js::jsval::{JSVal, NullValue, UndefinedValue};
use msg::constellation_msg::PipelineId;
use net_traits::{CoreResourceThread, FetchMetadata, FilteredMetadata};
use net_traits::{FetchMetadata, FilteredMetadata};
use net_traits::{FetchResponseListener, LoadOrigin, NetworkError, ReferrerPolicy};
use net_traits::CoreResourceMsg::Fetch;
use net_traits::request::{CredentialsMode, Destination, RequestInit, RequestMode};
use net_traits::trim_http_whitespace;
use network_listener::{NetworkListener, PreInvoke};
use script_runtime::ScriptChan;
use servo_atoms::Atom;
use std::ascii::AsciiExt;
use std::borrow::ToOwned;
@ -63,6 +62,7 @@ use std::cell::Cell;
use std::default::Default;
use std::str;
use std::sync::{Arc, Mutex};
use task_source::networking::NetworkingTaskSource;
use time;
use timers::{OneshotTimerCallback, OneshotTimerHandle};
use url::{Position, Url};
@ -214,8 +214,8 @@ impl XMLHttpRequest {
}
fn initiate_async_xhr(context: Arc<Mutex<XHRContext>>,
script_chan: Box<ScriptChan + Send>,
core_resource_thread: CoreResourceThread,
task_source: NetworkingTaskSource,
global: &GlobalScope,
init: RequestInit) {
impl FetchResponseListener for XHRContext {
fn process_request_body(&mut self) {
@ -262,13 +262,13 @@ impl XMLHttpRequest {
let (action_sender, action_receiver) = ipc::channel().unwrap();
let listener = NetworkListener {
context: context,
script_chan: script_chan,
wrapper: None,
task_source: task_source,
wrapper: Some(global.get_runnable_wrapper())
};
ROUTER.add_route(action_receiver.to_opaque(), box move |message| {
listener.notify_fetch(message.to().unwrap());
});
core_resource_thread.send(Fetch(init, action_sender)).unwrap();
global.core_resource_thread().send(Fetch(init, action_sender)).unwrap();
}
}
@ -1293,16 +1293,15 @@ impl XMLHttpRequest {
sync_status: DOMRefCell::new(None),
}));
let (script_chan, script_port) = if self.sync.get() {
let (task_source, script_port) = if self.sync.get() {
let (tx, rx) = global.new_script_pair();
(tx, Some(rx))
(NetworkingTaskSource(tx), Some(rx))
} else {
(global.networking_task_source(), None)
};
let core_resource_thread = global.core_resource_thread();
XMLHttpRequest::initiate_async_xhr(context.clone(), script_chan,
core_resource_thread, init);
XMLHttpRequest::initiate_async_xhr(context.clone(), task_source,
global, init);
if let Some(script_port) = script_port {
loop {

4
components/script/fetch.rs
View file

@ -97,8 +97,8 @@ pub fn Fetch(global: &GlobalScope, input: RequestInfo, init: &RequestInit) -> Rc
}));
let listener = NetworkListener {
context: fetch_context,
script_chan: global.networking_task_source(),
wrapper: None,
task_source: global.networking_task_source(),
wrapper: Some(global.get_runnable_wrapper())
};
ROUTER.add_route(action_receiver.to_opaque(), box move |message| {

10
components/script/network_listener.rs
View file

@ -4,16 +4,16 @@
use bluetooth_traits::{BluetoothResponseListener, BluetoothResponseResult};
use net_traits::{Action, FetchResponseListener, FetchResponseMsg};
use script_runtime::{CommonScriptMsg, ScriptChan};
use script_runtime::ScriptThreadEventCategory::NetworkEvent;
use script_thread::{Runnable, RunnableWrapper};
use std::sync::{Arc, Mutex};
use task_source::TaskSource;
use task_source::networking::NetworkingTaskSource;
/// An off-thread sink for async network event runnables. All such events are forwarded to
/// a target thread, where they are invoked on the provided context object.
pub struct NetworkListener<Listener: PreInvoke + Send + 'static> {
pub context: Arc<Mutex<Listener>>,
pub script_chan: Box<ScriptChan + Send>,
pub task_source: NetworkingTaskSource,
pub wrapper: Option<RunnableWrapper>,
}
@ -24,9 +24,9 @@ impl<Listener: PreInvoke + Send + 'static> NetworkListener<Listener> {
action: action,
};
let result = if let Some(ref wrapper) = self.wrapper {
self.script_chan.send(CommonScriptMsg::RunnableMsg(NetworkEvent, wrapper.wrap_runnable(runnable)))
self.task_source.queue_with_wrapper(runnable, wrapper)
} else {
self.script_chan.send(CommonScriptMsg::RunnableMsg(NetworkEvent, runnable))
self.task_source.queue_wrapperless(runnable)
};
if let Err(err) = result {
warn!("failed to deliver network data: {:?}", err);

7
components/script/script_thread.rs
View file

@ -659,7 +659,7 @@ impl ScriptThread {
chan: MainThreadScriptChan(chan.clone()),
dom_manipulation_task_source: DOMManipulationTaskSource(chan.clone()),
user_interaction_task_source: UserInteractionTaskSource(chan.clone()),
networking_task_source: NetworkingTaskSource(chan.clone()),
networking_task_source: NetworkingTaskSource(boxed_script_sender.clone()),
history_traversal_task_source: HistoryTraversalTaskSource(chan),
file_reading_task_source: FileReadingTaskSource(boxed_script_sender),
@ -1623,7 +1623,6 @@ impl ScriptThread {
let MainThreadScriptChan(ref sender) = self.chan;
let DOMManipulationTaskSource(ref dom_sender) = self.dom_manipulation_task_source;
let UserInteractionTaskSource(ref user_sender) = self.user_interaction_task_source;
let NetworkingTaskSource(ref network_sender) = self.networking_task_source;
let HistoryTraversalTaskSource(ref history_sender) = self.history_traversal_task_source;
let (ipc_timer_event_chan, ipc_timer_event_port) = ipc::channel().unwrap();
@ -1635,7 +1634,7 @@ impl ScriptThread {
MainThreadScriptChan(sender.clone()),
DOMManipulationTaskSource(dom_sender.clone()),
UserInteractionTaskSource(user_sender.clone()),
NetworkingTaskSource(network_sender.clone()),
self.networking_task_source.clone(),
HistoryTraversalTaskSource(history_sender.clone()),
self.file_reading_task_source.clone(),
self.image_cache_channel.clone(),
@ -2050,7 +2049,7 @@ impl ScriptThread {
let (action_sender, action_receiver) = ipc::channel().unwrap();
let listener = NetworkListener {
context: context,
script_chan: self.chan.clone(),
task_source: self.networking_task_source.clone(),
wrapper: None,
};
ROUTER.add_route(action_receiver.to_opaque(), box move |message| {

35
components/script/task_source/networking.rs
View file

@ -2,19 +2,32 @@
* License, v. 2.0. If a copy of the MPL was not distributed with this
* file, You can obtain one at http://mozilla.org/MPL/2.0/. */
use script_runtime::{CommonScriptMsg, ScriptChan};
use script_thread::MainThreadScriptMsg;
use std::sync::mpsc::Sender;
use script_runtime::{CommonScriptMsg, ScriptChan, ScriptThreadEventCategory};
use script_thread::{Runnable, RunnableWrapper};
use task_source::TaskSource;
#[derive(JSTraceable)]
pub struct NetworkingTaskSource(pub Sender<MainThreadScriptMsg>);
pub struct NetworkingTaskSource(pub Box<ScriptChan + Send + 'static>);
impl ScriptChan for NetworkingTaskSource {
fn send(&self, msg: CommonScriptMsg) -> Result<(), ()> {
self.0.send(MainThreadScriptMsg::Common(msg)).map_err(|_| ())
}
fn clone(&self) -> Box<ScriptChan + Send> {
box NetworkingTaskSource((&self.0).clone())
impl Clone for NetworkingTaskSource {
fn clone(&self) -> NetworkingTaskSource {
NetworkingTaskSource(self.0.clone())
}
}
impl TaskSource for NetworkingTaskSource {
fn queue_with_wrapper<T>(&self,
msg: Box<T>,
wrapper: &RunnableWrapper)
-> Result<(), ()>
where T: Runnable + Send + 'static {
self.0.send(CommonScriptMsg::RunnableMsg(ScriptThreadEventCategory::NetworkEvent,
wrapper.wrap_runnable(msg)))
}
}
impl NetworkingTaskSource {
pub fn queue_wrapperless<T: Runnable + Send + 'static>(&self, msg: Box<T>) -> Result<(), ()> {
self.0.send(CommonScriptMsg::RunnableMsg(ScriptThreadEventCategory::NetworkEvent, msg))
}
}