Added support for synchronization along with an unit test for cancellation listener

components/net/resource_task.rs

@@ -187,6 +187,9 @@ impl ResourceChannelManager {
                     }
                     self.resource_manager.cancel_load_map.remove(&res_id);
                 }
+                ControlMsg::Synchronize(sender) => {
+                    let _ = sender.send(());
+                }
                 ControlMsg::Exit => break,
             }
         }

components/net_traits/lib.rs

@@ -234,6 +234,8 @@ pub enum ControlMsg {
     GetCookiesForUrl(Url, IpcSender<Option<String>>, CookieSource),
     /// Cancel a network request corresponding to a given `ResourceId`
     Cancel(ResourceId),
+    /// Synchronization message solely for knowing the state of the ResourceChannelManager loop
+    Synchronize(IpcSender<()>),
     /// Break the load handler loop and exit
     Exit,
 }

tests/unit/net/resource_task.rs

@@ -170,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();
+}