Auto merge of #7844 - Wafflespeanut:requests, r=jdm

Cancelable network requests!

fixes #4974

<!-- Reviewable:start -->
[<img src="https://reviewable.io/review_button.png" height=40 alt="Review on Reviewable"/>](https://reviewable.io/reviews/servo/servo/7844)
<!-- Reviewable:end -->

tests/unit/net/data_loader.rs

@@ -18,13 +18,16 @@ fn assert_parse(url:          &'static str,
                 data:         Option<Vec<u8>>) {
     use net::data_loader::load;
     use net::mime_classifier::MIMEClassifier;
+    use net::resource_task::CancellationListener;
     use std::sync::Arc;
     use std::sync::mpsc::channel;
     use url::Url;
 
     let (start_chan, start_port) = ipc::channel().unwrap();
     let classifier = Arc::new(MIMEClassifier::new());
-    load(LoadData::new(Url::parse(url).unwrap(), None), Channel(start_chan), classifier);
+    load(LoadData::new(Url::parse(url).unwrap(), None),
+         Channel(start_chan),
+         classifier, CancellationListener::new(None));
 
     let response = start_port.recv().unwrap();
     assert_eq!(&response.metadata.content_type, &content_type);

tests/unit/net/resource_task.rs

@@ -11,7 +11,6 @@ use std::collections::HashMap;
 use std::sync::mpsc::channel;
 use url::Url;
 
-
 #[test]
 fn test_exit() {
     let resource_task = new_resource_task("".to_owned(), None);
@@ -23,7 +22,7 @@ fn test_bad_scheme() {
     let resource_task = new_resource_task("".to_owned(), None);
     let (start_chan, start) = ipc::channel().unwrap();
     let url = Url::parse("bogus://whatever").unwrap();
-    resource_task.send(ControlMsg::Load(LoadData::new(url, None), LoadConsumer::Channel(start_chan))).unwrap();
+    resource_task.send(ControlMsg::Load(LoadData::new(url, None), LoadConsumer::Channel(start_chan), None)).unwrap();
     let response = start.recv().unwrap();
     match response.progress_port.recv().unwrap() {
       ProgressMsg::Done(result) => { assert!(result.is_err()) }
@@ -171,3 +170,57 @@ fn test_replace_hosts() {
     let url = Url::parse("http://a.foo.bar.com").unwrap();
     assert_eq!(host_replacement(host_table, &url).domain().unwrap(), "a.foo.bar.com");
 }
+
+#[test]
+fn test_cancelled_listener() {
+    use std::io::Write;
+    use std::net::TcpListener;
+    use std::thread;
+
+    // http_loader always checks for headers in the response
+    let header = vec!["HTTP/1.1 200 OK",
+                      "Server: test-server",
+                      "Content-Type: text/plain",
+                      "\r\n"];
+    let body = vec!["Yay!", "We're doomed!"];
+
+    // Setup a TCP server to which requests are made
+    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    let port = listener.local_addr().unwrap().port();
+    let (body_sender, body_receiver) = channel();
+    thread::spawn(move || {
+        if let Ok((mut stream, _)) = listener.accept() {
+            // immediately stream the headers once the connection has been established
+            let _ = stream.write(header.join("\r\n").as_bytes());
+            // wait for the main thread to send the body, so as to ensure that we're
+            // doing everything sequentially
+            let body_vec: Vec<&str> = body_receiver.recv().unwrap();
+            let _ = stream.write(body_vec.join("\r\n").as_bytes());
+        }
+    });
+
+    let resource_task = new_resource_task("".to_owned(), None);
+    let (sender, receiver) = ipc::channel().unwrap();
+    let (id_sender, id_receiver) = ipc::channel().unwrap();
+    let (sync_sender, sync_receiver) = ipc::channel().unwrap();
+    let url = Url::parse(&format!("http://127.0.0.1:{}", port)).unwrap();
+
+    resource_task.send(ControlMsg::Load(LoadData::new(url, None),
+                                        LoadConsumer::Channel(sender),
+                                        Some(id_sender))).unwrap();
+    // get the `ResourceId` and send a cancel message, which should stop the loading loop
+    let res_id = id_receiver.recv().unwrap();
+    resource_task.send(ControlMsg::Cancel(res_id)).unwrap();
+    // synchronize with the resource_task loop, so that we don't simply send everything at once!
+    resource_task.send(ControlMsg::Synchronize(sync_sender)).unwrap();
+    let _ = sync_receiver.recv();
+    // now, let's send the body, because the connection is still active and data would be loaded
+    // (but, the loading has been cancelled)
+    let _ = body_sender.send(body);
+    let response = receiver.recv().unwrap();
+    match response.progress_port.recv().unwrap() {
+        ProgressMsg::Done(result) => assert_eq!(result.unwrap_err(), "load cancelled".to_owned()),
+        _ => panic!("baaaah!"),
+    }
+    resource_task.send(ControlMsg::Exit).unwrap();
+}