Create easy common interface for off-thread network listeners, and re…

…move the CORS-specific reimplementation of async networking.

components/net/resource_task.rs

@@ -54,8 +54,8 @@ pub fn global_init() {
 pub struct ListenerWrapper(pub Box<for<'r> Invoke<(&'r (AsyncResponseListener+'r))> + Send>);
 
 impl ListenerWrapper {
-    fn new<F>(f: Box<F>) -> ListenerWrapper
-              where F: for <'r> FnOnce(&'r (AsyncResponseListener+'r)) + Send {
+    pub fn new<F>(f: Box<F>) -> ListenerWrapper
+                  where F: for <'r> FnOnce(&'r (AsyncResponseListener+'r)) + Send {
         ListenerWrapper(f)
     }
 

components/script/cors.rs

@@ -9,9 +9,14 @@
 //! This library will eventually become the core of the Fetch crate
 //! with CORSRequest being expanded into FetchRequest (etc)
 
+use network_listener::{NetworkListener, PreInvoke};
+use script_task::ScriptChan;
+use net::resource_task::{AsyncResponseTarget, AsyncResponseListener, ListenerWrapper, Metadata};
+
 use std::ascii::AsciiExt;
 use std::borrow::ToOwned;
-use std::thunk::Invoke;
+use std::cell::RefCell;
+use std::sync::{Arc, Mutex};
 use time;
 use time::{now, Timespec};
 
@@ -28,27 +33,12 @@ use hyper::status::StatusClass::Success;
 use url::{SchemeData, Url};
 use util::task::spawn_named;
 
+/// Interface for network listeners concerned with CORS checks. Proper network requests
+/// should be initiated from this method, based on the response provided.
 pub trait AsyncCORSResponseListener {
     fn response_available(&self, response: CORSResponse);
 }
 
-pub trait AsyncCORSResponseTarget {
-    fn invoke_with_listener(&self, listener: CORSListenerWrapper);
-}
-
-pub struct CORSListenerWrapper(pub Box<for<'r> Invoke<(&'r (AsyncCORSResponseListener+'r))> + Send>);
-
-impl CORSListenerWrapper {
-    fn new<F>(f: Box<F>) -> CORSListenerWrapper
-              where F: for <'r> FnOnce(&'r (AsyncCORSResponseListener+'r)) + Send {
-        CORSListenerWrapper(f)
-    }
-
-    pub fn invoke(self, listener: &AsyncCORSResponseListener) {
-        (self.0).invoke(listener)
-    }
-}
-
 #[derive(Clone)]
 pub struct CORSRequest {
     pub origin: Url,
@@ -112,14 +102,47 @@ impl CORSRequest {
         }
     }
 
-    pub fn http_fetch_async(&self, listener: Box<AsyncCORSResponseTarget + Send>) {
+    pub fn http_fetch_async(&self,
+                            listener: Box<AsyncCORSResponseListener+Send>,
+                            script_chan: Box<ScriptChan+Send>) {
+        struct CORSContext {
+            listener: Box<AsyncCORSResponseListener+Send>,
+            response: RefCell<Option<CORSResponse>>,
+        }
+
+        // This is shoe-horning the CORSReponse stuff into the rest of the async network
+        // framework right now. It would be worth redesigning http_fetch to do this properly.
+        impl AsyncResponseListener for CORSContext {
+            fn headers_available(&self, _metadata: Metadata) {
+            }
+
+            fn data_available(&self, _payload: Vec<u8>) {
+            }
+
+            fn response_complete(&self, _status: Result<(), String>) {
+                let response = self.response.borrow_mut().take().unwrap();
+                self.listener.response_available(response);
+            }
+        }
+        impl PreInvoke for CORSContext {}
+
+        let context = CORSContext {
+            listener: listener,
+            response: RefCell::new(None),
+        };
+        let listener = NetworkListener {
+            context: Arc::new(Mutex::new(context)),
+            script_chan: script_chan,
+        };
+
         // TODO: this exists only to make preflight check non-blocking
         // perhaps should be handled by the resource task?
         let req = self.clone();
         spawn_named("cors".to_owned(), move || {
             let response = req.http_fetch();
-            listener.invoke_with_listener(CORSListenerWrapper::new(box move |listener: &AsyncCORSResponseListener| {
-                listener.response_available(response);
+            *listener.context.lock().unwrap().response.borrow_mut() = Some(response);
+            listener.invoke_with_listener(ListenerWrapper::new(box move |listener: &AsyncResponseListener| {
+                listener.response_complete(Ok(()));
             }));
         });
     }

components/script/dom/xmlhttprequest.rs

@@ -26,6 +26,7 @@ use dom::urlsearchparams::URLSearchParamsHelpers;
 use dom::xmlhttprequesteventtarget::XMLHttpRequestEventTarget;
 use dom::xmlhttprequesteventtarget::XMLHttpRequestEventTargetTypeId;
 use dom::xmlhttprequestupload::XMLHttpRequestUpload;
+use network_listener::{NetworkListener, PreInvoke};
 use script_task::{ScriptChan, ScriptMsg, Runnable, ScriptPort};
 
 use encoding::all::UTF_8;
@@ -43,12 +44,11 @@ use js::jsapi::JS_ClearPendingException;
 use js::jsval::{JSVal, NullValue, UndefinedValue};
 
 use net::resource_task::{ResourceTask, ResourceCORSData, LoadData, LoadConsumer, Metadata};
-use net::resource_task::{AsyncResponseListener, AsyncResponseTarget};
-use net::resource_task::ListenerWrapper;
+use net::resource_task::AsyncResponseListener;
 use net::resource_task::ControlMsg::Load;
 use net::resource_task::ProgressMsg::Done;
 use cors::{allow_cross_origin_request, CORSRequest, RequestMode, AsyncCORSResponseListener};
-use cors::{AsyncCORSResponseTarget, CORSListenerWrapper, CORSResponse};
+use cors::CORSResponse;
 use util::str::DOMString;
 use util::task::spawn_named;
 
@@ -224,44 +224,15 @@ impl XMLHttpRequest {
             }
         }
 
-        struct CORSListener {
-            context: Arc<Mutex<CORSContext>>,
-            script_chan: Box<ScriptChan+Send>,
-        }
-
-        impl AsyncCORSResponseTarget for CORSListener {
-            fn invoke_with_listener(&self, listener: CORSListenerWrapper) {
-                self.script_chan.send(ScriptMsg::RunnableMsg(box CORSRunnable {
-                    context: self.context.clone(),
-                    listener: listener,
-                }));
-            }
-        }
-
-        struct CORSRunnable {
-            context: Arc<Mutex<CORSContext>>,
-            listener: CORSListenerWrapper,
-        }
-
-        impl Runnable for CORSRunnable {
-            fn handler(self: Box<CORSRunnable>) {
-                let this = *self;
-                this.listener.invoke(&*this.context.lock().unwrap());
-            }
-        }
-
-        let cors_context = Arc::new(Mutex::new(CORSContext {
+        let cors_context = CORSContext {
             xhr: context,
             load_data: RefCell::new(Some(load_data)),
             req: req.clone(),
             script_chan: script_chan.clone(),
             resource_task: resource_task,
-        }));
+        };
 
-        req.http_fetch_async(box CORSListener {
-            context: cors_context,
-            script_chan: script_chan
-        });
+        req.http_fetch_async(box cors_context, script_chan);
     }
 
     fn initiate_async_xhr(context: Arc<Mutex<XHRContext>>,
@@ -292,40 +263,14 @@ impl XMLHttpRequest {
             }
         }
 
-        struct XHRRunnable {
-            context: Arc<Mutex<XHRContext>>,
-            listener: ListenerWrapper,
-        }
-
-        impl Runnable for XHRRunnable {
-            fn handler(self: Box<XHRRunnable>) {
-                let this = *self;
-
-                let context = this.context.lock().unwrap();
-                let xhr = context.xhr.to_temporary().root();
-                if xhr.r().generation_id.get() != context.gen_id {
-                    return;
-                }
-
-                this.listener.invoke(&*context);
-            }
-        }
-
-        struct XHRListener {
-            context: Arc<Mutex<XHRContext>>,
-            script_chan: Box<ScriptChan+Send>,
-        }
-
-        impl AsyncResponseTarget for XHRListener {
-            fn invoke_with_listener(&self, listener: ListenerWrapper) {
-                self.script_chan.send(ScriptMsg::RunnableMsg(box XHRRunnable {
-                    context: self.context.clone(),
-                    listener: listener,
-                }));
+        impl PreInvoke for XHRContext {
+            fn should_invoke(&self) -> bool {
+                let xhr = self.xhr.to_temporary().root();
+                xhr.r().generation_id.get() == self.gen_id
             }
         }
 
-        let listener = box XHRListener {
+        let listener = box NetworkListener {
             context: context,
             script_chan: script_chan,
         };
@@ -613,7 +558,7 @@ impl<'a> XMLHttpRequestMethods for JSRef<'a, XMLHttpRequest> {
         debug!("request_headers = {:?}", *self.request_headers.borrow());
 
         self.fetch_time.set(time::now().to_timespec().sec);
-        let rv = self.fetch2(load_data, cors_request, global.r());
+        let rv = self.fetch(load_data, cors_request, global.r());
         if self.sync.get() {
             return rv;
         }
@@ -752,8 +697,8 @@ trait PrivateXMLHttpRequestHelpers {
     fn cancel_timeout(self);
     fn filter_response_headers(self) -> Headers;
     fn discard_subsequent_responses(self);
-    fn fetch2(self, load_data: LoadData, cors_request: Result<Option<CORSRequest>,()>,
-              global: GlobalRef) -> ErrorResult;
+    fn fetch(self, load_data: LoadData, cors_request: Result<Option<CORSRequest>,()>,
+             global: GlobalRef) -> ErrorResult;
 }
 
 impl<'a> PrivateXMLHttpRequestHelpers for JSRef<'a, XMLHttpRequest> {
@@ -1070,8 +1015,7 @@ impl<'a> PrivateXMLHttpRequestHelpers for JSRef<'a, XMLHttpRequest> {
         self.response_status.set(Err(()));
     }
 
-    #[allow(unsafe_blocks)]
-    fn fetch2(self,
+    fn fetch(self,
               load_data: LoadData,
               cors_request: Result<Option<CORSRequest>,()>,
               global: GlobalRef) -> ErrorResult {

components/script/lib.rs

@@ -61,6 +61,7 @@ pub mod dom;
 pub mod parse;
 
 pub mod layout_interface;
+mod network_listener;
 pub mod page;
 pub mod script_task;
 mod timers;

components/script/network_listener.rs

@@ -0,0 +1,48 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * 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_task::{ScriptChan, ScriptMsg, Runnable};
+use net::resource_task::{AsyncResponseTarget, AsyncResponseListener, ListenerWrapper};
+use std::sync::{Arc, Mutex};
+
+/// 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<T: AsyncResponseListener + PreInvoke + Send> {
+    pub context: Arc<Mutex<T>>,
+    pub script_chan: Box<ScriptChan+Send>,
+}
+
+impl<T: AsyncResponseListener + PreInvoke + Send> AsyncResponseTarget for NetworkListener<T> {
+    fn invoke_with_listener(&self, listener: ListenerWrapper) {
+        self.script_chan.send(ScriptMsg::RunnableMsg(box ListenerRunnable {
+            context: self.context.clone(),
+            listener: listener,
+        }));
+    }
+}
+
+/// A gating mechanism that runs before invoking the runnable on the target thread.
+/// If the `should_invoke` method returns false, the runnable is discarded without
+/// being invoked.
+pub trait PreInvoke {
+    fn should_invoke(&self) -> bool {
+        true
+    }
+}
+
+/// A runnable for moving the async network events between threads.
+struct ListenerRunnable<T: AsyncResponseListener + PreInvoke + Send> {
+    context: Arc<Mutex<T>>,
+    listener: ListenerWrapper,
+}
+
+impl<T: AsyncResponseListener + PreInvoke + Send> Runnable for ListenerRunnable<T> {
+    fn handler(self: Box<ListenerRunnable<T>>) {
+        let this = *self;
+        let context = this.context.lock().unwrap();
+        if context.should_invoke() {
+            this.listener.invoke(&*context);
+        }
+    }
+}