Upgrade to latest Rust.

1 files changed, 189 insertions, 311 deletions

diff --git a/src/components/net/image_cache_task.rs b/src/components/net/image_cache_task.rs
index 7669f09f4d7..8a99ccf4a2c 100644
--- a/src/components/net/image_cache_task.rs
+++ b/src/components/net/image_cache_task.rs
@@ -7,8 +7,7 @@ use resource_task;
 use resource_task::ResourceTask;
 use servo_util::url::{UrlMap, url_map};
 
-use std::cell::Cell;
-use std::comm::{Chan, Port, SharedChan, stream};
+use std::comm::{Chan, Port, SharedChan};
 use std::task::spawn;
 use std::to_str::ToStr;
 use std::util::replace;
@@ -24,7 +23,7 @@ pub enum Msg {
     // FIXME: We can probably get rid of this Cell now
     // FIXME: make this priv after visibility rules change
     /// Used be the prefetch tasks to post back image binaries
-    StorePrefetchedImageData(Url, Result<Cell<~[u8]>, ()>),
+    StorePrefetchedImageData(Url, Result<~[u8], ()>),
 
     /// Tell the cache to decode an image. Must be posted before GetImage/WaitForImage
     Decode(Url),
@@ -40,12 +39,16 @@ pub enum Msg {
     /// Wait for an image to become available (or fail to load).
     WaitForImage(Url, Chan<ImageResponseMsg>),
 
+    /// Clients must wait for a response before shutting down the ResourceTask
+    Exit(Chan<()>),
+
     /// For testing
     // FIXME: make this priv after visibility rules change
-    OnMsg(~fn(msg: &Msg)),
+    WaitForStore(Chan<()>),
 
-    /// Clients must wait for a response before shutting down the ResourceTask
-    Exit(Chan<()>),
+    /// For testing
+    // FIXME: make this priv after visibility rules change
+    WaitForStorePrefetched(Chan<()>),
 }
 
 #[deriving(Clone)]
@@ -59,11 +62,11 @@ impl Eq for ImageResponseMsg {
     fn eq(&self, other: &ImageResponseMsg) -> bool {
         // FIXME: Bad copies
         match (self.clone(), other.clone()) {
-            (ImageReady(*), ImageReady(*)) => fail!(~"unimplemented comparison"),
+            (ImageReady(..), ImageReady(..)) => fail!(~"unimplemented comparison"),
             (ImageNotReady, ImageNotReady) => true,
             (ImageFailed, ImageFailed) => true,
 
-            (ImageReady(*), _) | (ImageNotReady, _) | (ImageFailed, _) => false
+            (ImageReady(..), _) | (ImageNotReady, _) | (ImageFailed, _) => false
         }
     }
 
@@ -72,49 +75,39 @@ impl Eq for ImageResponseMsg {
     }
 }
 
-pub type ImageCacheTask = SharedChan<Msg>;
-
-type DecoderFactory = ~fn() -> ~fn(&[u8]) -> Option<Image>;
-
-pub fn ImageCacheTask(resource_task: ResourceTask) -> ImageCacheTask {
-    ImageCacheTask_(resource_task, default_decoder_factory)
+#[deriving(Clone)]
+pub struct ImageCacheTask {
+    chan: SharedChan<Msg>,
 }
 
-pub fn ImageCacheTask_(resource_task: ResourceTask, decoder_factory: DecoderFactory)
-                       -> ImageCacheTask {
-    // FIXME: Doing some dancing to avoid copying decoder_factory, our test
-    // version of which contains an uncopyable type which rust will currently
-    // copy unsoundly
-    let decoder_factory_cell = Cell::new(decoder_factory);
+type DecoderFactory = fn() -> proc(&[u8]) -> Option<Image>;
 
-    let (port, chan) = stream();
-    let chan = SharedChan::new(chan);
-    let port_cell = Cell::new(port);
-    let chan_cell = Cell::new(chan.clone());
+pub fn ImageCacheTask(resource_task: ResourceTask) -> ImageCacheTask {
+    let (port, chan) = SharedChan::new();
+    let chan_clone = chan.clone();
 
-    do spawn {
+    spawn(proc() {
         let mut cache = ImageCache {
             resource_task: resource_task.clone(),
-            decoder_factory: decoder_factory_cell.take(),
-            port: port_cell.take(),
-            chan: chan_cell.take(),
+            port: port,
+            chan: chan_clone,
             state_map: url_map(),
             wait_map: url_map(),
             need_exit: None
         };
         cache.run();
-    }
+    });
 
-    chan
+    ImageCacheTask {
+        chan: chan,
+    }
 }
 
 // FIXME: make this priv after visibility rules change
 pub fn SyncImageCacheTask(resource_task: ResourceTask) -> ImageCacheTask {
-    let (port, chan) = stream();
-    let port_cell = Cell::new(port);
+    let (port, chan) = SharedChan::new();
 
-    do spawn {
-        let port = port_cell.take();
+    spawn(proc() {
         let inner_cache = ImageCacheTask(resource_task.clone());
 
         loop {
@@ -131,16 +124,16 @@ pub fn SyncImageCacheTask(resource_task: ResourceTask) -> ImageCacheTask {
                 msg => inner_cache.send(msg)
             }
         }
-    }
+    });
 
-    SharedChan::new(chan)
+    ImageCacheTask {
+        chan: chan,
+    }
 }
 
 struct ImageCache {
     /// A handle to the resource task for fetching the image binaries
     resource_task: ResourceTask,
-    /// Creates image decoders
-    decoder_factory: DecoderFactory,
     /// The port on which we'll receive client requests
     port: Port<Msg>,
     /// A copy of the shared chan to give to child tasks
@@ -156,7 +149,7 @@ struct ImageCache {
 enum ImageState {
     Init,
     Prefetching(AfterPrefetch),
-    Prefetched(Cell<~[u8]>),
+    Prefetched(~[u8]),
     Decoding,
     Decoded(Arc<~Image>),
     Failed
@@ -170,29 +163,39 @@ enum AfterPrefetch {
 
 impl ImageCache {
     pub fn run(&mut self) {
-        let mut msg_handlers: ~[~fn(msg: &Msg)] = ~[];
+        let mut store_chan: Option<Chan<()>> = None;
+        let mut store_prefetched_chan: Option<Chan<()>> = None;
 
         loop {
             let msg = self.port.recv();
 
-            for handler in msg_handlers.iter() {
-                (*handler)(&msg)
-            }
-
             debug!("image_cache_task: received: {:?}", msg);
 
             match msg {
                 Prefetch(url) => self.prefetch(url),
                 StorePrefetchedImageData(url, data) => {
+                    store_prefetched_chan.map(|chan| {
+                        chan.send(());
+                    });
+                    store_prefetched_chan = None;
+
                     self.store_prefetched_image_data(url, data);
                 }
                 Decode(url) => self.decode(url),
-                StoreImage(url, image) => self.store_image(url, image),
+                StoreImage(url, image) => {
+                    store_chan.map(|chan| {
+                        chan.send(());
+                    });
+                    store_chan = None;
+
+                    self.store_image(url, image)
+                }
                 GetImage(url, response) => self.get_image(url, response),
                 WaitForImage(url, response) => {
                     self.wait_for_image(url, response)
                 }
-                OnMsg(handler) => msg_handlers.push(handler),
+                WaitForStore(chan) => store_chan = Some(chan),
+                WaitForStorePrefetched(chan) => store_prefetched_chan = Some(chan),
                 Exit(response) => {
                     assert!(self.need_exit.is_none());
                     self.need_exit = Some(response);
@@ -208,10 +211,10 @@ impl ImageCache {
                 let mut can_exit = true;
                 for (_, state) in self.state_map.iter() {
                     match *state {
-                        Prefetching(*) => can_exit = false,
+                        Prefetching(..) => can_exit = false,
                         Decoding => can_exit = false,
 
-                        Init | Prefetched(*) | Decoded(*) | Failed => ()
+                        Init | Prefetched(..) | Decoded(..) | Failed => ()
                     }
                 }
 
@@ -243,45 +246,44 @@ impl ImageCache {
             Init => {
                 let to_cache = self.chan.clone();
                 let resource_task = self.resource_task.clone();
-                let url_cell = Cell::new(url.clone());
+                let url_clone = url.clone();
 
-                do spawn {
-                    let url = url_cell.take();
+                spawn(proc() {
+                    let url = url_clone;
                     debug!("image_cache_task: started fetch for {:s}", url.to_str());
 
                     let image = load_image_data(url.clone(), resource_task.clone());
 
                     let result = if image.is_ok() {
-                        Ok(Cell::new(image.unwrap()))
+                        Ok(image.unwrap())
                     } else {
                         Err(())
                     };
                     to_cache.send(StorePrefetchedImageData(url.clone(), result));
                     debug!("image_cache_task: ended fetch for {:s}", (url.clone()).to_str());
-                }
+                });
 
                 self.set_state(url, Prefetching(DoNotDecode));
             }
 
-            Prefetching(*) | Prefetched(*) | Decoding | Decoded(*) | Failed => {
+            Prefetching(..) | Prefetched(..) | Decoding | Decoded(..) | Failed => {
                 // We've already begun working on this image
             }
         }
     }
 
-    fn store_prefetched_image_data(&mut self, url: Url, data: Result<Cell<~[u8]>, ()>) {
+    fn store_prefetched_image_data(&mut self, url: Url, data: Result<~[u8], ()>) {
         match self.get_state(url.clone()) {
           Prefetching(next_step) => {
             match data {
-              Ok(data_cell) => {
-                let data = data_cell.take();
-                self.set_state(url.clone(), Prefetched(Cell::new(data)));
+              Ok(data) => {
+                self.set_state(url.clone(), Prefetched(data));
                 match next_step {
                   DoDecode => self.decode(url),
                   _ => ()
                 }
               }
-              Err(*) => {
+              Err(..) => {
                 self.set_state(url.clone(), Failed);
                 self.purge_waiters(url, || ImageFailed);
               }
@@ -289,9 +291,9 @@ impl ImageCache {
           }
 
           Init
-          | Prefetched(*)
+          | Prefetched(..)
           | Decoding
-          | Decoded(*)
+          | Decoded(..)
           | Failed => {
             fail!(~"wrong state for storing prefetched image")
           }
@@ -311,18 +313,14 @@ impl ImageCache {
                 // We don't have the data yet, but the decode request is queued up
             }
 
-            Prefetched(data_cell) => {
-                assert!(!data_cell.is_empty());
-
-                let data = data_cell.take();
+            Prefetched(data) => {
                 let to_cache = self.chan.clone();
-                let url_cell = Cell::new(url.clone());
-                let decode = (self.decoder_factory)();
+                let url_clone = url.clone();
 
-                do spawn {
-                    let url = url_cell.take();
+                spawn(proc() {
+                    let url = url_clone;
                     debug!("image_cache_task: started image decode for {:s}", url.to_str());
-                    let image = decode(data);
+                    let image = load_from_memory(data);
                     let image = if image.is_some() {
                         Some(Arc::new(~image.unwrap()))
                     } else {
@@ -330,12 +328,12 @@ impl ImageCache {
                     };
                     to_cache.send(StoreImage(url.clone(), image));
                     debug!("image_cache_task: ended image decode for {:s}", url.to_str());
-                }
+                });
 
                 self.set_state(url, Decoding);
             }
 
-            Decoding | Decoded(*) | Failed => {
+            Decoding | Decoded(..) | Failed => {
                 // We've already begun decoding
             }
         }
@@ -358,9 +356,9 @@ impl ImageCache {
           }
 
           Init
-          | Prefetching(*)
-          | Prefetched(*)
-          | Decoded(*)
+          | Prefetching(..)
+          | Prefetched(..)
+          | Decoded(..)
           | Failed => {
             fail!(~"incorrect state in store_image")
           }
@@ -368,7 +366,7 @@ impl ImageCache {
 
     }
 
-    fn purge_waiters(&mut self, url: Url, f: &fn() -> ImageResponseMsg) {
+    fn purge_waiters(&mut self, url: Url, f: || -> ImageResponseMsg) {
         match self.wait_map.pop(&url) {
             Some(waiters) => {
                 unsafe {
@@ -387,7 +385,7 @@ impl ImageCache {
         match self.get_state(url.clone()) {
             Init => fail!(~"request for image before prefetch"),
             Prefetching(DoDecode) => response.send(ImageNotReady),
-            Prefetching(DoNotDecode) | Prefetched(*) => fail!(~"request for image before decode"),
+            Prefetching(DoNotDecode) | Prefetched(..) => fail!(~"request for image before decode"),
             Decoding => response.send(ImageNotReady),
             Decoded(image) => response.send(ImageReady(image.clone())),
             Failed => response.send(ImageFailed),
@@ -398,16 +396,16 @@ impl ImageCache {
         match self.get_state(url.clone()) {
             Init => fail!(~"request for image before prefetch"),
 
-            Prefetching(DoNotDecode) | Prefetched(*) => fail!(~"request for image before decode"),
+            Prefetching(DoNotDecode) | Prefetched(..) => fail!(~"request for image before decode"),
 
             Prefetching(DoDecode) | Decoding => {
                 // We don't have this image yet
                 if self.wait_map.contains_key(&url) {
                     let waiters = self.wait_map.find_mut(&url).unwrap();
+                    let mut response = Some(response);
                     unsafe {
-                        let res = Cell::new(response);
-                        waiters.unsafe_access( |waiters| {
-                            waiters.push(res.take());
+                        waiters.unsafe_access(|waiters| {
+                            waiters.push(response.take().unwrap());
                         });
                     }
                 } else {
@@ -434,14 +432,34 @@ trait ImageCacheTaskClient {
 
 impl ImageCacheTaskClient for ImageCacheTask {
     fn exit(&self) {
-        let (response_port, response_chan) = stream();
+        let (response_port, response_chan) = Chan::new();
         self.send(Exit(response_chan));
         response_port.recv();
     }
 }
 
+impl ImageCacheTask {
+    pub fn send(&self, msg: Msg) {
+        self.chan.send(msg);
+    }
+    
+    #[cfg(test)]
+    fn wait_for_store(&self) -> Port<()> {
+        let (port, chan) = Chan::new();
+        self.send(WaitForStore(chan));
+        port
+    }
+
+    #[cfg(test)]
+    fn wait_for_store_prefetched(&self) -> Port<()> {
+        let (port, chan) = Chan::new();
+        self.send(WaitForStorePrefetched(chan));
+        port
+    }
+}
+
 fn load_image_data(url: Url, resource_task: ResourceTask) -> Result<~[u8], ()> {
-    let (response_port, response_chan) = stream();
+    let (response_port, response_chan) = Chan::new();
     resource_task.send(resource_task::Load(url, response_chan));
 
     let mut image_data = ~[];
@@ -452,38 +470,29 @@ fn load_image_data(url: Url, resource_task: ResourceTask) -> Result<~[u8], ()> {
             resource_task::Payload(data) => {
                 image_data.push_all(data);
             }
-            resource_task::Done(result::Ok(*)) => {
+            resource_task::Done(result::Ok(..)) => {
                 return Ok(image_data);
             }
-            resource_task::Done(result::Err(*)) => {
+            resource_task::Done(result::Err(..)) => {
                 return Err(());
             }
         }
     }
 }
 
-fn default_decoder_factory() -> ~fn(&[u8]) -> Option<Image> {
-    let foo: ~fn(&[u8]) -> Option<Image> = |data: &[u8]| { load_from_memory(data) };
-    foo
-}
 
 #[cfg(test)]
 mod tests {
     use super::*;
 
-    use std::comm;
-    use std::comm::{Port, SharedChan};
-    use std::result;
-    use std::cell::Cell;
-
     use resource_task;
     use resource_task::{ResourceTask, Metadata, start_sending};
-    use image::base::{Image, test_image_bin, load_from_memory};
+    use image::base::test_image_bin;
     use util::spawn_listener;
     use servo_util::url::make_url;
 
-    fn mock_resource_task(on_load: ~fn(resource: Chan<resource_task::ProgressMsg>)) -> ResourceTask {
-        let chan = do spawn_listener |port: Port<resource_task::ControlMsg>| {
+    fn mock_resource_task(on_load: proc(resource: SharedChan<resource_task::ProgressMsg>)) -> ResourceTask {
+        spawn_listener(proc(port: Port<resource_task::ControlMsg>) {
             loop {
                 match port.recv() {
                     resource_task::Load(_, response) => {
@@ -493,13 +502,12 @@ mod tests {
                     resource_task::Exit => break
                 }
             }
-        };
-        SharedChan::new(chan)
+        })
     }
 
     #[test]
     fn should_exit_on_request() {
-        let mock_resource_task = mock_resource_task(|_response| () );
+        let mock_resource_task = mock_resource_task(proc(_response) {});
 
         let image_cache_task = ImageCacheTask(mock_resource_task.clone());
         let _url = make_url(~"file", None);
@@ -511,24 +519,24 @@ mod tests {
     #[test]
     #[should_fail]
     fn should_fail_if_unprefetched_image_is_requested() {
-        let mock_resource_task = mock_resource_task(|_response| () );
+        let mock_resource_task = mock_resource_task(proc(_response) {});
 
         let image_cache_task = ImageCacheTask(mock_resource_task.clone());
         let url = make_url(~"file", None);
 
-        let (port, chan) = stream();
+        let (port, chan) = Chan::new();
         image_cache_task.send(GetImage(url, chan));
         port.recv();
     }
 
     #[test]
     fn should_request_url_from_resource_task_on_prefetch() {
-        let (url_requested, url_requested_chan) = comm::stream();
+        let (url_requested, url_requested_chan) = Chan::new();
 
-        let mock_resource_task = do mock_resource_task |response| {
+        let mock_resource_task = mock_resource_task(proc(response) {
             url_requested_chan.send(());
-            response.send(resource_task::Done(result::Ok(())));
-        };
+            response.send(resource_task::Done(Ok(())));
+        });
 
         let image_cache_task = ImageCacheTask(mock_resource_task.clone());
         let url = make_url(~"file", None);
@@ -539,47 +547,14 @@ mod tests {
         mock_resource_task.send(resource_task::Exit);
     }
 
-
-    #[test]
-    #[should_fail]
-    fn should_fail_if_requesting_decode_of_an_unprefetched_image() {
-        let mock_resource_task = mock_resource_task(|_response| () );
-
-        let image_cache_task = ImageCacheTask(mock_resource_task.clone());
-        let url = make_url(~"file", None);
-
-        image_cache_task.send(Decode(url));
-        image_cache_task.exit();
-    }
-
-    #[test]
-    #[should_fail]
-    fn should_fail_if_requesting_image_before_requesting_decode() {
-        let mock_resource_task = do mock_resource_task |response| {
-            response.send(resource_task::Done(result::Ok(())));
-        };
-
-        let image_cache_task = ImageCacheTask(mock_resource_task.clone());
-        let url = make_url(~"file", None);
-
-        image_cache_task.send(Prefetch(url.clone()));
-        // no decode message
-
-        let (_port, chan) = stream();
-        image_cache_task.send(GetImage(url, chan));
-
-        image_cache_task.exit();
-        mock_resource_task.send(resource_task::Exit);
-    }
-
     #[test]
     fn should_not_request_url_from_resource_task_on_multiple_prefetches() {
-        let (url_requested, url_requested_chan) = comm::stream();
+        let (url_requested, url_requested_chan) = Chan::new();
 
-        let mock_resource_task = do mock_resource_task |response| {
+        let mock_resource_task = mock_resource_task(proc(response) {
             url_requested_chan.send(());
-            response.send(resource_task::Done(result::Ok(())));
-        };
+            response.send(resource_task::Done(Ok(())));
+        });
 
         let image_cache_task = ImageCacheTask(mock_resource_task.clone());
         let url = make_url(~"file", None);
@@ -589,27 +564,27 @@ mod tests {
         url_requested.recv();
         image_cache_task.exit();
         mock_resource_task.send(resource_task::Exit);
-        assert!(!url_requested.peek())
+        assert!(url_requested.try_recv().is_none())
     }
 
     #[test]
     fn should_return_image_not_ready_if_data_has_not_arrived() {
-        let (wait_port, wait_chan) = comm::stream();
+        let (wait_port, wait_chan) = Chan::new();
 
-        let mock_resource_task = do mock_resource_task |response| {
+        let mock_resource_task = mock_resource_task(proc(response) {
             // Don't send the data until after the client requests
             // the image
             wait_port.recv();
             response.send(resource_task::Payload(test_image_bin()));
-            response.send(resource_task::Done(result::Ok(())));
-        };
+            response.send(resource_task::Done(Ok(())));
+        });
 
         let image_cache_task = ImageCacheTask(mock_resource_task.clone());
         let url = make_url(~"file", None);
 
         image_cache_task.send(Prefetch(url.clone()));
         image_cache_task.send(Decode(url.clone()));
-        let (response_port, response_chan) = stream();
+        let (response_port, response_chan) = Chan::new();
         image_cache_task.send(GetImage(url, response_chan));
         assert!(response_port.recv() == ImageNotReady);
         wait_chan.send(());
@@ -619,30 +594,23 @@ mod tests {
 
     #[test]
     fn should_return_decoded_image_data_if_data_has_arrived() {
-        let mock_resource_task = do mock_resource_task |response| {
+        let mock_resource_task = mock_resource_task(proc(response) {
             response.send(resource_task::Payload(test_image_bin()));
-            response.send(resource_task::Done(result::Ok(())));
-        };
+            response.send(resource_task::Done(Ok(())));
+        });
 
         let image_cache_task = ImageCacheTask(mock_resource_task.clone());
         let url = make_url(~"file", None);
 
-        let (wait_for_image, wait_for_image_chan) = comm::stream();
-
-        image_cache_task.send(OnMsg(|msg| {
-            match *msg {
-              StoreImage(*) => wait_for_image_chan.send(()),
-              _ => ()
-            }
-        }));
+        let join_port = image_cache_task.wait_for_store();
 
         image_cache_task.send(Prefetch(url.clone()));
         image_cache_task.send(Decode(url.clone()));
 
         // Wait until our mock resource task has sent the image to the image cache
-        wait_for_image.recv();
+        join_port.recv();
 
-        let (response_port, response_chan) = stream();
+        let (response_port, response_chan) = Chan::new();
         image_cache_task.send(GetImage(url, response_chan));
         match response_port.recv() {
           ImageReady(_) => (),
@@ -655,31 +623,24 @@ mod tests {
 
     #[test]
     fn should_return_decoded_image_data_for_multiple_requests() {
-        let mock_resource_task = do mock_resource_task |response| {
+        let mock_resource_task = mock_resource_task(proc(response) {
             response.send(resource_task::Payload(test_image_bin()));
-            response.send(resource_task::Done(result::Ok(())));
-        };
+            response.send(resource_task::Done(Ok(())));
+        });
 
         let image_cache_task = ImageCacheTask(mock_resource_task.clone());
         let url = make_url(~"file", None);
 
-        let (wait_for_image, wait_for_image_chan) = comm::stream();
-
-        image_cache_task.send(OnMsg(|msg| {
-            match *msg {
-              StoreImage(*) => wait_for_image_chan.send(()),
-              _ => ()
-            }
-        }));
+        let join_port = image_cache_task.wait_for_store();
 
         image_cache_task.send(Prefetch(url.clone()));
         image_cache_task.send(Decode(url.clone()));
 
         // Wait until our mock resource task has sent the image to the image cache
-        wait_for_image.recv();
+        join_port.recv();
 
         for _ in range(0,2) {
-            let (response_port, response_chan) = stream();
+            let (response_port, response_chan) = Chan::new();
             image_cache_task.send(GetImage(url.clone(), response_chan));
             match response_port.recv() {
               ImageReady(_) => (),
@@ -693,17 +654,17 @@ mod tests {
 
     #[test]
     fn should_not_request_image_from_resource_task_if_image_is_already_available() {
-        let (image_bin_sent, image_bin_sent_chan) = comm::stream();
+        let (image_bin_sent, image_bin_sent_chan) = Chan::new();
 
-        let (resource_task_exited, resource_task_exited_chan) = comm::stream();
+        let (resource_task_exited, resource_task_exited_chan) = Chan::new();
 
-        let mock_resource_task = do spawn_listener |port: comm::Port<resource_task::ControlMsg>| {
+        let mock_resource_task = spawn_listener(proc(port: Port<resource_task::ControlMsg>) {
             loop {
                 match port.recv() {
                     resource_task::Load(_, response) => {
                         let chan = start_sending(response, Metadata::default(make_url(~"file:///fake", None)));
                         chan.send(resource_task::Payload(test_image_bin()));
-                        chan.send(resource_task::Done(result::Ok(())));
+                        chan.send(resource_task::Done(Ok(())));
                         image_bin_sent_chan.send(());
                     }
                     resource_task::Exit => {
@@ -712,8 +673,7 @@ mod tests {
                     }
                 }
             }
-        };
-        let mock_resource_task = SharedChan::new(mock_resource_task);
+        });
 
         let image_cache_task = ImageCacheTask(mock_resource_task.clone());
         let url = make_url(~"file", None);
@@ -732,22 +692,22 @@ mod tests {
 
         // Our resource task should not have received another request for the image
         // because it's already cached
-        assert!(!image_bin_sent.peek());
+        assert!(image_bin_sent.try_recv().is_none());
     }
 
     #[test]
     fn should_not_request_image_from_resource_task_if_image_fetch_already_failed() {
-        let (image_bin_sent, image_bin_sent_chan) = comm::stream();
+        let (image_bin_sent, image_bin_sent_chan) = Chan::new();
 
-        let (resource_task_exited, resource_task_exited_chan) = comm::stream();
+        let (resource_task_exited, resource_task_exited_chan) = Chan::new();
 
-        let mock_resource_task = do spawn_listener |port: comm::Port<resource_task::ControlMsg>| {
+        let mock_resource_task = spawn_listener(proc(port: Port<resource_task::ControlMsg>) {
             loop {
                 match port.recv() {
                     resource_task::Load(_, response) => {
                         let chan = start_sending(response, Metadata::default(make_url(~"file:///fake", None)));
                         chan.send(resource_task::Payload(test_image_bin()));
-                        chan.send(resource_task::Done(result::Err(())));
+                        chan.send(resource_task::Done(Err(())));
                         image_bin_sent_chan.send(());
                     }
                     resource_task::Exit => {
@@ -756,8 +716,7 @@ mod tests {
                     }
                 }
             }
-        };
-        let mock_resource_task = SharedChan::new(mock_resource_task);
+        });
 
         let image_cache_task = ImageCacheTask(mock_resource_task.clone());
         let url = make_url(~"file", None);
@@ -778,36 +737,29 @@ mod tests {
 
         // Our resource task should not have received another request for the image
         // because it's already cached
-        assert!(!image_bin_sent.peek());
+        assert!(image_bin_sent.try_recv().is_none());
     }
 
     #[test]
     fn should_return_failed_if_image_bin_cannot_be_fetched() {
-        let mock_resource_task = do mock_resource_task |response| {
+        let mock_resource_task = mock_resource_task(proc(response) {
             response.send(resource_task::Payload(test_image_bin()));
             // ERROR fetching image
-            response.send(resource_task::Done(result::Err(())));
-        };
+            response.send(resource_task::Done(Err(())));
+        });
 
         let image_cache_task = ImageCacheTask(mock_resource_task.clone());
         let url = make_url(~"file", None);
 
-        let (wait_for_prefetech, wait_for_prefetech_chan) = comm::stream();
-
-        image_cache_task.send(OnMsg(|msg| {
-            match *msg {
-              StorePrefetchedImageData(*) => wait_for_prefetech_chan.send(()),
-              _ => ()
-            }
-        }));
+        let join_port = image_cache_task.wait_for_store_prefetched();
 
         image_cache_task.send(Prefetch(url.clone()));
         image_cache_task.send(Decode(url.clone()));
 
         // Wait until our mock resource task has sent the image to the image cache
-        wait_for_prefetech.recv();
+        join_port.recv();
 
-        let (response_port, response_chan) = stream();
+        let (response_port, response_chan) = Chan::new();
         image_cache_task.send(GetImage(url, response_chan));
         match response_port.recv() {
           ImageFailed => (),
@@ -820,31 +772,24 @@ mod tests {
 
     #[test]
     fn should_return_failed_for_multiple_get_image_requests_if_image_bin_cannot_be_fetched() {
-        let mock_resource_task = do mock_resource_task |response | {
+        let mock_resource_task = mock_resource_task(proc(response) {
             response.send(resource_task::Payload(test_image_bin()));
             // ERROR fetching image
-            response.send(resource_task::Done(result::Err(())));
-        };
+            response.send(resource_task::Done(Err(())));
+        });
 
         let image_cache_task = ImageCacheTask(mock_resource_task.clone());
         let url = make_url(~"file", None);
 
-        let (wait_for_prefetech, wait_for_prefetech_chan) = comm::stream();
-
-        image_cache_task.send(OnMsg(|msg| {
-            match *msg {
-              StorePrefetchedImageData(*) => wait_for_prefetech_chan.send(()),
-              _ => ()
-            }
-        }));
+        let join_port = image_cache_task.wait_for_store_prefetched();
 
         image_cache_task.send(Prefetch(url.clone()));
         image_cache_task.send(Decode(url.clone()));
 
         // Wait until our mock resource task has sent the image to the image cache
-        wait_for_prefetech.recv();
+        join_port.recv();
 
-        let (response_port, response_chan) = stream();
+        let (response_port, response_chan) = Chan::new();
         image_cache_task.send(GetImage(url.clone(), response_chan));
         match response_port.recv() {
           ImageFailed => (),
@@ -852,7 +797,7 @@ mod tests {
         }
 
         // And ask again, we should get the same response
-        let (response_port, response_chan) = stream();
+        let (response_port, response_chan) = Chan::new();
         image_cache_task.send(GetImage(url, response_chan));
         match response_port.recv() {
           ImageFailed => (),
@@ -864,86 +809,26 @@ mod tests {
     }
 
     #[test]
-    fn should_return_not_ready_if_image_is_still_decoding() {
-        let (wait_to_decode_port, wait_to_decode_chan) = comm::stream();
-
-        let mock_resource_task = do mock_resource_task |response| {
-            response.send(resource_task::Payload(test_image_bin()));
-            response.send(resource_task::Done(result::Ok(())));
-        };
-
-        let wait_to_decode_port_cell = Cell::new(wait_to_decode_port);
-        let decoder_factory: ~fn:Send() -> ~fn:Send(&[u8]) -> Option<Image> = || {
-            let wait_to_decode_port = wait_to_decode_port_cell.take();
-            |data: &[u8]| {
-                // Don't decode until after the client requests the image
-                wait_to_decode_port.recv();
-                load_from_memory(data)
-            }
-        };
-
-        let image_cache_task = ImageCacheTask_(mock_resource_task.clone(), decoder_factory);
-        let url = make_url(~"file", None);
-
-        let (wait_for_prefetech, wait_for_prefetech_chan) = comm::stream();
-
-        image_cache_task.send(OnMsg(|msg| {
-            match *msg {
-              StorePrefetchedImageData(*) => wait_for_prefetech_chan.send(()),
-              _ => ()
-            }
-        }));
-
-        image_cache_task.send(Prefetch(url.clone()));
-        image_cache_task.send(Decode(url.clone()));
-
-        // Wait until our mock resource task has sent the image to the image cache
-        wait_for_prefetech.recv();
-
-        // Make the request
-        let (response_port, response_chan) = stream();
-        image_cache_task.send(GetImage(url, response_chan));
-
-        match response_port.recv() {
-          ImageNotReady => (),
-          _ => fail!("bleh")
-        }
-
-        // Now decode
-        wait_to_decode_chan.send(());
-
-        image_cache_task.exit();
-        mock_resource_task.send(resource_task::Exit);
-    }
-
-    #[test]
     fn should_return_failed_if_image_decode_fails() {
-        let mock_resource_task = do mock_resource_task |response| {
+        let mock_resource_task = mock_resource_task(proc(response) {
             // Bogus data
             response.send(resource_task::Payload(~[]));
-            response.send(resource_task::Done(result::Ok(())));
-        };
+            response.send(resource_task::Done(Ok(())));
+        });
 
         let image_cache_task = ImageCacheTask(mock_resource_task.clone());
         let url = make_url(~"file", None);
 
-        let (wait_for_decode, wait_for_decode_chan) = comm::stream();
-
-        image_cache_task.send(OnMsg(|msg| {
-            match *msg {
-              StoreImage(*) => wait_for_decode_chan.send(()),
-              _ => ()
-            }
-        }));
+        let join_port = image_cache_task.wait_for_store();
 
         image_cache_task.send(Prefetch(url.clone()));
         image_cache_task.send(Decode(url.clone()));
 
         // Wait until our mock resource task has sent the image to the image cache
-        wait_for_decode.recv();
+        join_port.recv();
 
         // Make the request
-        let (response_port, response_chan) = stream();
+        let (response_port, response_chan) = Chan::new();
         image_cache_task.send(GetImage(url, response_chan));
 
         match response_port.recv() {
@@ -957,33 +842,26 @@ mod tests {
 
     #[test]
     fn should_return_image_on_wait_if_image_is_already_loaded() {
-        let mock_resource_task = do mock_resource_task |response| {
+        let mock_resource_task = mock_resource_task(proc(response) {
             response.send(resource_task::Payload(test_image_bin()));
-            response.send(resource_task::Done(result::Ok(())));
-        };
+            response.send(resource_task::Done(Ok(())));
+        });
 
         let image_cache_task = ImageCacheTask(mock_resource_task.clone());
         let url = make_url(~"file", None);
 
-        let (wait_for_decode, wait_for_decode_chan) = comm::stream();
-
-        image_cache_task.send(OnMsg(|msg| {
-            match *msg {
-              StoreImage(*) => wait_for_decode_chan.send(()),
-              _ => ()
-            }
-        }));
+        let join_port = image_cache_task.wait_for_store();
 
         image_cache_task.send(Prefetch(url.clone()));
         image_cache_task.send(Decode(url.clone()));
 
         // Wait until our mock resource task has sent the image to the image cache
-        wait_for_decode.recv();
+        join_port.recv();
 
-        let (response_port, response_chan) = stream();
+        let (response_port, response_chan) = Chan::new();
         image_cache_task.send(WaitForImage(url, response_chan));
         match response_port.recv() {
-          ImageReady(*) => (),
+          ImageReady(..) => (),
           _ => fail!("bleh")
         }
 
@@ -993,13 +871,13 @@ mod tests {
 
     #[test]
     fn should_return_image_on_wait_if_image_is_not_yet_loaded() {
-        let (wait_port, wait_chan) = comm::stream();
+        let (wait_port, wait_chan) = Chan::new();
 
-        let mock_resource_task = do mock_resource_task |response| {
+        let mock_resource_task = mock_resource_task(proc(response) {
             wait_port.recv();
             response.send(resource_task::Payload(test_image_bin()));
-            response.send(resource_task::Done(result::Ok(())));
-        };
+            response.send(resource_task::Done(Ok(())));
+        });
 
         let image_cache_task = ImageCacheTask(mock_resource_task.clone());
         let url = make_url(~"file", None);
@@ -1007,13 +885,13 @@ mod tests {
         image_cache_task.send(Prefetch(url.clone()));
         image_cache_task.send(Decode(url.clone()));
 
-        let (response_port, response_chan) = stream();
+        let (response_port, response_chan) = Chan::new();
         image_cache_task.send(WaitForImage(url, response_chan));
 
         wait_chan.send(());
 
         match response_port.recv() {
-          ImageReady(*) => (),
+          ImageReady(..) => (),
           _ => fail!("bleh")
         }
 
@@ -1023,13 +901,13 @@ mod tests {
 
     #[test]
     fn should_return_image_failed_on_wait_if_image_fails_to_load() {
-        let (wait_port, wait_chan) = comm::stream();
+        let (wait_port, wait_chan) = Chan::new();
 
-        let mock_resource_task = do mock_resource_task |response| {
+        let mock_resource_task = mock_resource_task(proc(response) {
             wait_port.recv();
             response.send(resource_task::Payload(test_image_bin()));
-            response.send(resource_task::Done(result::Err(())));
-        };
+            response.send(resource_task::Done(Err(())));
+        });
 
         let image_cache_task = ImageCacheTask(mock_resource_task.clone());
         let url = make_url(~"file", None);
@@ -1037,7 +915,7 @@ mod tests {
         image_cache_task.send(Prefetch(url.clone()));
         image_cache_task.send(Decode(url.clone()));
 
-        let (response_port, response_chan) = stream();
+        let (response_port, response_chan) = Chan::new();
         image_cache_task.send(WaitForImage(url, response_chan));
 
         wait_chan.send(());
@@ -1053,10 +931,10 @@ mod tests {
 
     #[test]
     fn sync_cache_should_wait_for_images() {
-        let mock_resource_task = do mock_resource_task |response| {
+        let mock_resource_task = mock_resource_task(proc(response) {
             response.send(resource_task::Payload(test_image_bin()));
-            response.send(resource_task::Done(result::Ok(())));
-        };
+            response.send(resource_task::Done(Ok(())));
+        });
 
         let image_cache_task = SyncImageCacheTask(mock_resource_task.clone());
         let url = make_url(~"file", None);
@@ -1064,7 +942,7 @@ mod tests {
         image_cache_task.send(Prefetch(url.clone()));
         image_cache_task.send(Decode(url.clone()));
 
-        let (response_port, response_chan) = stream();
+        let (response_port, response_chan) = Chan::new();
         image_cache_task.send(GetImage(url, response_chan));
         match response_port.recv() {
           ImageReady(_) => (),