Completely abstract sending responses over channels vs. listeners.

components/net/file_loader.rs

@@ -2,17 +2,16 @@
  * 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 net_traits::{LoadData, Metadata, ProgressMsg, ResponseSenders};
+use net_traits::{LoadData, Metadata, ResponseSenders};
 use net_traits::ProgressMsg::{Payload, Done};
 use mime_classifier::MIMEClassifier;
-use resource_task::{start_sending, start_sending_sniffed};
+use resource_task::{start_sending, start_sending_sniffed, ProgressSender};
 
 use std::borrow::ToOwned;
 use std::fs::File;
 use std::io::Read;
 use std::path::PathBuf;
 use std::sync::Arc;
-use std::sync::mpsc::Sender;
 use util::task::spawn_named;
 
 static READ_SIZE: usize = 8192;
@@ -34,7 +33,7 @@ fn read_block(reader: &mut File) -> Result<ReadStatus, String> {
     }
 }
 
-fn read_all(reader: &mut File, progress_chan: &Sender<ProgressMsg>)
+fn read_all(reader: &mut File, progress_chan: &ProgressSender)
             -> Result<(), String> {
     loop {
         match try!(read_block(reader)) {

components/net/resource_task.rs

@@ -13,7 +13,7 @@ use cookie;
 use mime_classifier::MIMEClassifier;
 
 use net_traits::{ControlMsg, LoadData, LoadResponse, ResponseSenders, LoadConsumer};
-use net_traits::{Metadata, ProgressMsg, ResourceTask};
+use net_traits::{Metadata, ProgressMsg, ResourceTask, AsyncResponseTarget, ResponseAction};
 use net_traits::ProgressMsg::Done;
 use util::opts;
 use util::task::spawn_named;
@@ -58,22 +58,44 @@ pub fn global_init() {
     }
 }
 
+pub enum ProgressSender {
+    Channel(Sender<ProgressMsg>),
+    Listener(Box<AsyncResponseTarget>),
+}
+
+impl ProgressSender {
+    //XXXjdm return actual error
+    pub fn send(&self, msg: ProgressMsg) -> Result<(), ()> {
+        match *self {
+            ProgressSender::Channel(ref c) => c.send(msg).map_err(|_| ()),
+            ProgressSender::Listener(ref b) => {
+                let action = match msg {
+                    ProgressMsg::Payload(buf) => ResponseAction::DataAvailable(buf),
+                    ProgressMsg::Done(status) => ResponseAction::ResponseComplete(status),
+                };
+                b.invoke_with_listener(action);
+                Ok(())
+            }
+        }
+    }
+}
+
 /// For use by loaders in responding to a Load message.
-pub fn start_sending(start_chan: ResponseSenders, metadata: Metadata) -> Sender<ProgressMsg> {
+pub fn start_sending(start_chan: ResponseSenders, metadata: Metadata) -> ProgressSender {
     start_sending_opt(start_chan, metadata).ok().unwrap()
 }
 
 /// For use by loaders in responding to a Load message that allows content sniffing.
 pub fn start_sending_sniffed(start_chan: ResponseSenders, metadata: Metadata,
                              classifier: Arc<MIMEClassifier>, partial_body: &Vec<u8>)
-                             -> Sender<ProgressMsg> {
+                             -> ProgressSender {
     start_sending_sniffed_opt(start_chan, metadata, classifier, partial_body).ok().unwrap()
 }
 
 /// For use by loaders in responding to a Load message that allows content sniffing.
 pub fn start_sending_sniffed_opt(start_chan: ResponseSenders, mut metadata: Metadata,
                                  classifier: Arc<MIMEClassifier>, partial_body: &Vec<u8>)
-                                 -> Result<Sender<ProgressMsg>, ()> {
+                                 -> Result<ProgressSender, ()> {
     if opts::get().sniff_mime_types {
         // TODO: should be calculated in the resource loader, from pull requeset #4094
         let nosniff = false;
@@ -94,7 +116,7 @@ pub fn start_sending_sniffed_opt(start_chan: ResponseSenders, mut metadata: Meta
 }
 
 /// For use by loaders in responding to a Load message.
-pub fn start_sending_opt(start_chan: ResponseSenders, metadata: Metadata) -> Result<Sender<ProgressMsg>, ()> {
+pub fn start_sending_opt(start_chan: ResponseSenders, metadata: Metadata) -> Result<ProgressSender, ()> {
     match start_chan {
         ResponseSenders::Channel(start_chan) => {
             let (progress_chan, progress_port) = channel();
@@ -103,11 +125,14 @@ pub fn start_sending_opt(start_chan: ResponseSenders, metadata: Metadata) -> Res
                 progress_port: progress_port,
             });
             match result {
-                Ok(_) => Ok(progress_chan),
+                Ok(_) => Ok(ProgressSender::Channel(progress_chan)),
                 Err(_) => Err(())
             }
         }
-        ResponseSenders::Listener(_) => panic!(),
+        ResponseSenders::Listener(target) => {
+            target.invoke_with_listener(ResponseAction::HeadersAvailable(metadata));
+            Ok(ProgressSender::Listener(target))
+        }
     }
 }
 

components/net_traits/lib.rs

@@ -72,8 +72,24 @@ pub trait AsyncResponseListener {
     fn response_complete(&self, status: Result<(), String>);
 }
 
+pub enum ResponseAction {
+    HeadersAvailable(Metadata),
+    DataAvailable(Vec<u8>),
+    ResponseComplete(Result<(), String>)
+}
+
+impl ResponseAction {
+    pub fn process(self, listener: &AsyncResponseListener) {
+        match self {
+            ResponseAction::HeadersAvailable(m) => listener.headers_available(m),
+            ResponseAction::DataAvailable(d) => listener.data_available(d),
+            ResponseAction::ResponseComplete(r) => listener.response_complete(r),
+        }
+    }
+}
+
 pub trait AsyncResponseTarget {
-    fn get_listener(&self) -> &AsyncResponseListener;
+    fn invoke_with_listener(&self, action: ResponseAction);
 }
 
 pub enum LoadConsumer {
@@ -174,7 +190,7 @@ pub enum CookieSource {
 
 pub enum ResponseSenders {
     Channel(Sender<LoadResponse>),
-    Listener(Box<AsyncResponseTarget + Send>),
+    Listener(Box<AsyncResponseTarget+ Send>),
 }
 
 impl ResponseSenders {

tests/unit/net/image_cache_task.rs

@@ -6,7 +6,7 @@ use net::image_cache_task::*;
 use net_traits::image_cache_task::ImageResponseMsg::*;
 use net_traits::image_cache_task::Msg::*;
 
-use net::resource_task::start_sending;
+use net::resource_task::{start_sending, ProgressSender};
 use net_traits::{ControlMsg, Metadata, ProgressMsg, ResourceTask, ResponseSenders};
 use net_traits::image_cache_task::{ImageCacheTask, ImageCacheTaskClient, ImageResponseMsg, Msg};
 use net_traits::ProgressMsg::{Payload, Done};
@@ -41,7 +41,7 @@ impl ImageCacheTaskHelper for ImageCacheTask {
 }
 
 trait Closure {
-    fn invoke(&self, _response: Sender<ProgressMsg>) { }
+    fn invoke(&self, _response: ProgressSender) { }
 }
 struct DoesNothing;
 impl Closure for DoesNothing { }
@@ -50,7 +50,7 @@ struct JustSendOK {
     url_requested_chan: Sender<()>,
 }
 impl Closure for JustSendOK {
-    fn invoke(&self, response: Sender<ProgressMsg>) {
+    fn invoke(&self, response: ProgressSender) {
         self.url_requested_chan.send(()).unwrap();
         response.send(Done(Ok(()))).unwrap();
     }
@@ -58,7 +58,7 @@ impl Closure for JustSendOK {
 
 struct SendTestImage;
 impl Closure for SendTestImage {
-    fn invoke(&self, response: Sender<ProgressMsg>) {
+    fn invoke(&self, response: ProgressSender) {
         response.send(Payload(test_image_bin())).unwrap();
         response.send(Done(Ok(()))).unwrap();
     }
@@ -66,7 +66,7 @@ impl Closure for SendTestImage {
 
 struct SendBogusImage;
 impl Closure for SendBogusImage {
-    fn invoke(&self, response: Sender<ProgressMsg>) {
+    fn invoke(&self, response: ProgressSender) {
         response.send(Payload(vec!())).unwrap();
         response.send(Done(Ok(()))).unwrap();
     }
@@ -74,7 +74,7 @@ impl Closure for SendBogusImage {
 
 struct SendTestImageErr;
 impl Closure for SendTestImageErr {
-    fn invoke(&self, response: Sender<ProgressMsg>) {
+    fn invoke(&self, response: ProgressSender) {
         response.send(Payload(test_image_bin())).unwrap();
         response.send(Done(Err("".to_string()))).unwrap();
     }
@@ -84,7 +84,7 @@ struct WaitSendTestImage {
     wait_port: Receiver<()>,
 }
 impl Closure for WaitSendTestImage {
-    fn invoke(&self, response: Sender<ProgressMsg>) {
+    fn invoke(&self, response: ProgressSender) {
         // Don't send the data until after the client requests
         // the image
         self.wait_port.recv().unwrap();
@@ -97,7 +97,7 @@ struct WaitSendTestImageErr {
     wait_port: Receiver<()>,
 }
 impl Closure for WaitSendTestImageErr {
-    fn invoke(&self, response: Sender<ProgressMsg>) {
+    fn invoke(&self, response: ProgressSender) {
         // Don't send the data until after the client requests
         // the image
         self.wait_port.recv().unwrap();