Shared Sniffer Task

- Added TargetedLoadResponse and ResponseSenders
- LoadData constructor contains the next consumer which means
  SnifferManager doesn't need the next consumer to start
- New SnifferTask is created at new resource_task creation
- Update Unit Tests

components/net/about_loader.rs

@@ -2,7 +2,7 @@
  * 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 resource_task::{LoadResponse, Metadata, Done, LoadData, start_sending};
+use resource_task::{TargetedLoadResponse, Metadata, Done, LoadData, start_sending, ResponseSenders};
 use file_loader;
 
 use std::io::fs::PathExtensions;
@@ -11,10 +11,14 @@ use http::status::Ok as StatusOk;
 use servo_util::resource_files::resources_dir_path;
 
 
-pub fn factory(mut load_data: LoadData, start_chan: Sender<LoadResponse>) {
+pub fn factory(mut load_data: LoadData, start_chan: Sender<TargetedLoadResponse>) {
+    let senders = ResponseSenders {
+        immediate_consumer: start_chan.clone(),
+        eventual_consumer: load_data.consumer.clone(),
+    };
     match load_data.url.non_relative_scheme_data().unwrap() {
         "blank" => {
-            let chan = start_sending(start_chan, Metadata {
+            let chan = start_sending(senders, Metadata {
                 final_url: load_data.url,
                 content_type: Some(("text".to_string(), "html".to_string())),
                 charset: Some("utf-8".to_string()),
@@ -32,7 +36,7 @@ pub fn factory(mut load_data: LoadData, start_chan: Sender<LoadResponse>) {
             load_data.url = Url::from_file_path(&path).unwrap();
         }
         _ => {
-            start_sending(start_chan, Metadata::default(load_data.url))
+            start_sending(senders, Metadata::default(load_data.url))
                 .send(Done(Err("Unknown about: URL.".to_string())));
             return
         }

components/net/data_loader.rs

@@ -2,7 +2,7 @@
  * 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 resource_task::{Done, Payload, Metadata, LoadData, LoadResponse, start_sending};
+use resource_task::{Done, Payload, Metadata, LoadData, TargetedLoadResponse, start_sending, ResponseSenders};
 
 use serialize::base64::FromBase64;
 
@@ -11,20 +11,25 @@ use http::headers::content_type::MediaType;
 use url::{percent_decode, NonRelativeSchemeData};
 
 
-pub fn factory(load_data: LoadData, start_chan: Sender<LoadResponse>) {
+pub fn factory(load_data: LoadData, start_chan: Sender<TargetedLoadResponse>) {
     // NB: we don't spawn a new task.
     // Hypothesis: data URLs are too small for parallel base64 etc. to be worth it.
     // Should be tested at some point.
     // Left in separate function to allow easy moving to a task, if desired.
     load(load_data, start_chan)
 }
 
-fn load(load_data: LoadData, start_chan: Sender<LoadResponse>) {
+fn load(load_data: LoadData, start_chan: Sender<TargetedLoadResponse>) {
     let url = load_data.url;
     assert!("data" == url.scheme.as_slice());
 
     let mut metadata = Metadata::default(url.clone());
 
+    let senders = ResponseSenders {
+        immediate_consumer: start_chan,
+        eventual_consumer: load_data.consumer,
+    };
+
     // Split out content type and data.
     let mut scheme_data = match url.scheme_data {
         NonRelativeSchemeData(scheme_data) => scheme_data,
@@ -39,7 +44,7 @@ fn load(load_data: LoadData, start_chan: Sender<LoadResponse>) {
     }
     let parts: Vec<&str> = scheme_data.as_slice().splitn(1, ',').collect();
     if parts.len() != 2 {
-        start_sending(start_chan, metadata).send(Done(Err("invalid data uri".to_string())));
+        start_sending(senders, metadata).send(Done(Err("invalid data uri".to_string())));
         return;
     }
 
@@ -57,7 +62,7 @@ fn load(load_data: LoadData, start_chan: Sender<LoadResponse>) {
     let content_type: Option<MediaType> = from_stream_with_str(ct_str);
     metadata.set_content_type(&content_type);
 
-    let progress_chan = start_sending(start_chan, metadata);
+    let progress_chan = start_sending(senders, metadata);
     let bytes = percent_decode(parts[1].as_bytes());
 
     if is_base64 {
@@ -86,9 +91,11 @@ fn assert_parse(url:          &'static str,
                 data:         Option<Vec<u8>>) {
     use std::comm;
     use url::Url;
+    use sniffer_task;
 
     let (start_chan, start_port) = comm::channel();
-    load(LoadData::new(Url::parse(url).unwrap()), start_chan);
+    let sniffer_task = sniffer_task::new_sniffer_task();
+    load(LoadData::new(Url::parse(url).unwrap(), start_chan), sniffer_task);
 
     let response = start_port.recv();
     assert_eq!(&response.metadata.content_type, &content_type);

components/net/file_loader.rs

@@ -2,8 +2,7 @@
  * 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 resource_task::{ProgressMsg, Metadata, Payload, Done, LoadData, start_sending};
-use resource_task::{LoadResponse};
+use resource_task::{ProgressMsg, Metadata, Payload, Done, LoadData, start_sending, TargetedLoadResponse, ResponseSenders};
 
 use std::io;
 use std::io::File;
@@ -30,10 +29,14 @@ fn read_all(reader: &mut io::Stream, progress_chan: &Sender<ProgressMsg>)
     }
 }
 
-pub fn factory(load_data: LoadData, start_chan: Sender<LoadResponse>) {
+pub fn factory(load_data: LoadData, start_chan: Sender<TargetedLoadResponse>) {
     let url = load_data.url;
     assert!("file" == url.scheme.as_slice());
-    let progress_chan = start_sending(start_chan, Metadata::default(url.clone()));
+    let senders = ResponseSenders {
+        immediate_consumer: start_chan,
+        eventual_consumer: load_data.consumer,
+    };
+    let progress_chan = start_sending(senders, Metadata::default(url.clone()));
     spawn_named("file_loader", proc() {
         let file_path: Result<Path, ()> = url.to_file_path();
         match file_path {

components/net/http_loader.rs

@@ -2,7 +2,7 @@
  * 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 resource_task::{Metadata, Payload, Done, LoadResponse, LoadData, start_sending_opt};
+use resource_task::{Metadata, Payload, Done, TargetedLoadResponse, LoadData, start_sending_opt, ResponseSenders};
 
 use log;
 use std::collections::HashSet;
@@ -12,21 +12,21 @@ use std::io::Reader;
 use servo_util::task::spawn_named;
 use url::Url;
 
-pub fn factory(load_data: LoadData, start_chan: Sender<LoadResponse>) {
+pub fn factory(load_data: LoadData, start_chan: Sender<TargetedLoadResponse>) {
     spawn_named("http_loader", proc() load(load_data, start_chan))
 }
 
-fn send_error(url: Url, err: String, start_chan: Sender<LoadResponse>) {
+fn send_error(url: Url, err: String, senders: ResponseSenders) {
     let mut metadata = Metadata::default(url);
     metadata.status = None;
 
-    match start_sending_opt(start_chan, metadata) {
+    match start_sending_opt(senders, metadata) {
         Ok(p) => p.send(Done(Err(err))),
         _ => {}
     };
 }
 
-fn load(load_data: LoadData, start_chan: Sender<LoadResponse>) {
+fn load(load_data: LoadData, start_chan: Sender<TargetedLoadResponse>) {
     // FIXME: At the time of writing this FIXME, servo didn't have any central
     //        location for configuration. If you're reading this and such a
     //        repository DOES exist, please update this constant to use it.
@@ -35,17 +35,22 @@ fn load(load_data: LoadData, start_chan: Sender<LoadResponse>) {
     let mut url = load_data.url.clone();
     let mut redirected_to = HashSet::new();
 
+    let senders = ResponseSenders {
+        immediate_consumer: start_chan,
+        eventual_consumer: load_data.consumer
+    };
+
     // Loop to handle redirects.
     loop {
         iters = iters + 1;
 
         if iters > max_redirects {
-            send_error(url, "too many redirects".to_string(), start_chan);
+            send_error(url, "too many redirects".to_string(), senders);
             return;
         }
 
         if redirected_to.contains(&url) {
-            send_error(url, "redirect loop".to_string(), start_chan);
+            send_error(url, "redirect loop".to_string(), senders);
             return;
         }
 
@@ -55,7 +60,7 @@ fn load(load_data: LoadData, start_chan: Sender<LoadResponse>) {
             "http" | "https" => {}
             _ => {
                 let s = format!("{:s} request, but we don't support that scheme", url.scheme);
-                send_error(url, s, start_chan);
+                send_error(url, s, senders);
                 return;
             }
         }
@@ -66,7 +71,7 @@ fn load(load_data: LoadData, start_chan: Sender<LoadResponse>) {
         let mut writer = match request {
             Ok(w) => box w,
             Err(e) => {
-                send_error(url, e.desc.to_string(), start_chan);
+                send_error(url, e.desc.to_string(), senders);
                 return;
             }
         };
@@ -84,7 +89,7 @@ fn load(load_data: LoadData, start_chan: Sender<LoadResponse>) {
                 writer.headers.content_length = Some(data.len());
                 match writer.write(data.as_slice()) {
                     Err(e) => {
-                        send_error(url, e.desc.to_string(), start_chan);
+                        send_error(url, e.desc.to_string(), senders);
                         return;
                     }
                     _ => {}
@@ -95,7 +100,7 @@ fn load(load_data: LoadData, start_chan: Sender<LoadResponse>) {
         let mut response = match writer.read_response() {
             Ok(r) => r,
             Err((_, e)) => {
-                send_error(url, e.desc.to_string(), start_chan);
+                send_error(url, e.desc.to_string(), senders);
                 return;
             }
         };
@@ -116,7 +121,7 @@ fn load(load_data: LoadData, start_chan: Sender<LoadResponse>) {
                         Some(ref c) => {
                             if c.preflight {
                                 // The preflight lied
-                                send_error(url, "Preflight fetch inconsistent with main fetch".to_string(), start_chan);
+                                send_error(url, "Preflight fetch inconsistent with main fetch".to_string(), senders);
                                 return;
                             } else {
                                 // XXXManishearth There are some CORS-related steps here,
@@ -138,7 +143,7 @@ fn load(load_data: LoadData, start_chan: Sender<LoadResponse>) {
         metadata.headers = Some(response.headers.clone());
         metadata.status = Some(response.status.clone());
 
-        let progress_chan = match start_sending_opt(start_chan, metadata) {
+        let progress_chan = match start_sending_opt(senders, metadata) {
             Ok(p) => p,
             _ => return
         };

components/net/image_cache_task.rs

@@ -443,7 +443,7 @@ impl ImageCacheTask {
 
 fn load_image_data(url: Url, resource_task: ResourceTask) -> Result<Vec<u8>, ()> {
     let (response_chan, response_port) = channel();
-    resource_task.send(resource_task::Load(LoadData::new(url), response_chan));
+    resource_task.send(resource_task::Load(LoadData::new(url, response_chan)));
 
     let mut image_data = vec!();
 
@@ -481,7 +481,8 @@ mod tests {
     use super::*;
 
     use resource_task;
-    use resource_task::{ResourceTask, Metadata, start_sending};
+    use resource_task::{ResourceTask, Metadata, start_sending, ResponseSenders};
+    use sniffer_task;
     use image::base::test_image_bin;
     use servo_util::taskpool::TaskPool;
     use std::comm;
@@ -557,8 +558,13 @@ mod tests {
         spawn_listener(proc(port: Receiver<resource_task::ControlMsg>) {
             loop {
                 match port.recv() {
-                    resource_task::Load(_, response) => {
-                        let chan = start_sending(response, Metadata::default(
+                    resource_task::Load(response) => {
+                        let sniffer_task = sniffer_task::new_sniffer_task();
+                        let senders = ResponseSenders {
+                            immediate_consumer: sniffer_task,
+                            eventual_consumer: response.consumer.clone(),
+                        };
+                        let chan = start_sending(senders, Metadata::default(
                             Url::parse("file:///fake").unwrap()));
                         on_load.invoke(chan);
                     }
@@ -708,8 +714,13 @@ mod tests {
         let mock_resource_task = spawn_listener(proc(port: Receiver<resource_task::ControlMsg>) {
             loop {
                 match port.recv() {
-                    resource_task::Load(_, response) => {
-                        let chan = start_sending(response, Metadata::default(
+                    resource_task::Load(response) => {
+                        let sniffer_task = sniffer_task::new_sniffer_task();
+                        let senders = ResponseSenders {
+                            immediate_consumer: sniffer_task,
+                            eventual_consumer: response.consumer.clone(),
+                        };
+                        let chan = start_sending(senders, Metadata::default(
                             Url::parse("file:///fake").unwrap()));
                         chan.send(resource_task::Payload(test_image_bin()));
                         chan.send(resource_task::Done(Ok(())));
@@ -755,8 +766,13 @@ mod tests {
         let mock_resource_task = spawn_listener(proc(port: Receiver<resource_task::ControlMsg>) {
             loop {
                 match port.recv() {
-                    resource_task::Load(_, response) => {
-                        let chan = start_sending(response, Metadata::default(
+                    resource_task::Load(response) => {
+                        let sniffer_task = sniffer_task::new_sniffer_task();
+                        let senders = ResponseSenders {
+                            immediate_consumer: sniffer_task,
+                            eventual_consumer: response.consumer.clone(),
+                        };
+                        let chan = start_sending(senders, Metadata::default(
                             Url::parse("file:///fake").unwrap()));
                         chan.send(resource_task::Payload(test_image_bin()));
                         chan.send(resource_task::Done(Err("".to_string())));