Skip to content

Commit

Permalink
set response.body asynchronously in Fetch
Browse files Browse the repository at this point in the history
  • Loading branch information
@nikkisquared
nikkisquared committed Mar 7, 2016
1 parent 22ce878 commit b187985
Show file tree
Hide file tree
Showing 5 changed files with 110 additions and 69 deletions.
140 changes: 86 additions & 54 deletions components/net/fetch/methods.rs
View file
Expand Up @@ -5,7 +5,7 @@
use fetch::cors_cache::{BasicCORSCache, CORSCache, CacheRequestDetails};
use fetch::response::ResponseMethods;
use http_loader::{NetworkHttpRequestFactory, WrappedHttpResponse};
use http_loader::{create_http_connector, obtain_response};
use http_loader::{create_http_connector, obtain_response, read_block, ReadResult};
use hyper::client::response::Response as HyperResponse;
use hyper::header::{Accept, CacheControl, IfMatch, IfRange, IfUnmodifiedSince, Location};
use hyper::header::{AcceptLanguage, ContentLength, ContentLanguage, HeaderView, Pragma};
Expand All @@ -27,6 +27,7 @@ use std::cell::RefCell;
use std::io::Read;
use std::rc::Rc;
use std::str::FromStr;
use std::sync::{Arc, Mutex};
use std::thread;
use url::idna::domain_to_ascii;
use url::{Origin as UrlOrigin, OpaqueOrigin, Url, UrlParser, whatwg_scheme_type_mapper};
Expand All @@ -35,8 +36,9 @@ use util::thread::spawn_named;
pub fn fetch_async(request: Request, listener: Box<AsyncFetchListener + Send>) {
spawn_named(format!("fetch for {:?}", request.current_url_string()), move || {
let request = Rc::new(request);
let res = fetch(request);
listener.response_available(res);
let fetch_response = fetch(request);
fetch_response.wait_until_done();
listener.response_available(fetch_response);
})
}

Expand Down Expand Up @@ -140,9 +142,7 @@ fn main_fetch(request: Rc<Request>, cors_flag: bool, recursive_flag: bool) -> Re
// TODO this step

// Step 8
if !request.synchronous && !recursive_flag {
// TODO run the remaining steps in parallel
}
// this step is obsoleted by fetch_async

// Step 9
let mut response = if response.is_none() {
Expand Down Expand Up @@ -228,17 +228,18 @@ fn main_fetch(request: Rc<Request>, cors_flag: bool, recursive_flag: bool) -> Re
Method::Head | Method::Connect => true,
_ => false })
{
// when the Fetch implementation does asynchronous retrieval of the body,
// we will need to make sure nothing tries to write to the body at this point
*internal_response.body.borrow_mut() = ResponseBody::Empty;
// when Fetch is used only asynchronously, we will need to make sure
// that nothing tries to write to the body at this point
let mut body = internal_response.body.lock().unwrap();
*body = ResponseBody::Empty;
}

// Step 15
// TODO be able to compare response integrity against request integrity metadata
// if !response.is_network_error() {

// // Substep 1
// // TODO wait for response
// response.wait_until_done();

// // Substep 2
// if response.termination_reason.is_none() {
Expand All @@ -250,7 +251,7 @@ fn main_fetch(request: Rc<Request>, cors_flag: bool, recursive_flag: bool) -> Re

// Step 16
if request.synchronous {
// TODO wait for internal_response
response.get_actual_response().wait_until_done();
return response;
}

Expand All @@ -274,22 +275,14 @@ fn main_fetch(request: Rc<Request>, cors_flag: bool, recursive_flag: bool) -> Re
// Step 18
// TODO this step

match *internal_response.body.borrow() {
// Step 20
ResponseBody::Empty => {
// Substep 1
// Substep 2
},
// Step 19
internal_response.wait_until_done();

// Step 19
_ => {
// Substep 1
// Substep 2
}
};
// Step 20
// TODO this step
}

// TODO remove this line when asynchronous fetches are supported
// TODO remove this line when only asynchronous fetches are used
return response;
}

Expand Down Expand Up @@ -544,11 +537,12 @@ fn http_redirect_fetch(request: Rc<Request>,
let location = match response.get_actual_response().headers.get::<Location>() {
Some(&Location(ref location)) => location.clone(),
// Step 4
_ => return Response::network_error(),
_ => return Response::network_error()
};

// Step 5
let location_url = UrlParser::new().base_url(&request.current_url()).parse(&*location);
let response_url = response.get_actual_response().url.as_ref().unwrap();
let location_url = UrlParser::new().base_url(response_url).parse(&*location);

// Step 6
let location_url = match location_url {
Expand Down Expand Up @@ -663,29 +657,38 @@ fn http_network_or_cache_fetch(request: Rc<Request>,
http_request.headers.borrow_mut().set(UserAgent(global_user_agent().to_owned()));
}

// Step 9
if http_request.cache_mode.get() == CacheMode::Default && is_no_store_cache(&http_request.headers.borrow()) {
http_request.cache_mode.set(CacheMode::NoStore);
}
match http_request.cache_mode.get() {

// Step 10
if http_request.cache_mode.get() == CacheMode::Reload {
// Step 9
CacheMode::Default if is_no_store_cache(&http_request.headers.borrow()) => {
http_request.cache_mode.set(CacheMode::NoStore);
},

// Substep 1
if !http_request.headers.borrow().has::<Pragma>() {
http_request.headers.borrow_mut().set(Pragma::NoCache);
}
// Step 10
CacheMode::NoCache if !http_request.headers.borrow().has::<CacheControl>() => {
http_request.headers.borrow_mut().set(CacheControl(vec![CacheDirective::MaxAge(0)]));
},

// Substep 2
if !http_request.headers.borrow().has::<CacheControl>() {
http_request.headers.borrow_mut().set(CacheControl(vec![CacheDirective::NoCache]));
}
// Step 11
CacheMode::Reload => {
// Substep 1
if !http_request.headers.borrow().has::<Pragma>() {
http_request.headers.borrow_mut().set(Pragma::NoCache);
}

// Substep 2
if !http_request.headers.borrow().has::<CacheControl>() {
http_request.headers.borrow_mut().set(CacheControl(vec![CacheDirective::NoCache]));
}
},

_ => {}
}

// Step 11
// Step 12
// modify_request_headers(http_request.headers.borrow());

// Step 12
// Step 13
// TODO some of this step can't be implemented yet
if credentials_flag {

Expand Down Expand Up @@ -723,13 +726,13 @@ fn http_network_or_cache_fetch(request: Rc<Request>,
}
}

// Step 13
// TODO this step can't be implemented

// Step 14
let mut response: Option<Response> = None;
// TODO this step can't be implemented yet

// Step 15
let mut response: Option<Response> = None;

// Step 16
// TODO have a HTTP cache to check for a completed response
let complete_http_response_from_cache: Option<Response> = None;
if http_request.cache_mode.get() != CacheMode::NoStore &&
Expand Down Expand Up @@ -761,20 +764,20 @@ fn http_network_or_cache_fetch(request: Rc<Request>,
// TODO this substep
}

// Step 16
// Step 17
// TODO have a HTTP cache to check for a partial response
} else if http_request.cache_mode.get() == CacheMode::Default ||
http_request.cache_mode.get() == CacheMode::ForceCache {
// TODO this substep
}

// Step 17
// Step 18
if response.is_none() {
response = Some(http_network_fetch(request.clone(), http_request.clone(), credentials_flag));
}
let response = response.unwrap();

// Step 18
// Step 19
if let Some(status) = response.status {
if status == StatusCode::NotModified &&
(http_request.cache_mode.get() == CacheMode::Default ||
Expand All @@ -800,7 +803,7 @@ fn http_network_or_cache_fetch(request: Rc<Request>,
}
}

// Step 19
// Step 20
response
}

Expand Down Expand Up @@ -835,14 +838,43 @@ fn http_network_fetch(request: Rc<Request>,
let mut response = Response::new();
match wrapped_response {
Ok(mut res) => {
// is it okay for res.version to be unused?
response.url = Some(res.response.url.clone());
response.status = Some(res.response.status);
response.headers = res.response.headers.clone();

let mut body = vec![];
res.response.read_to_end(&mut body);
*response.body.borrow_mut() = ResponseBody::Done(body);
let res_body = response.body.clone();
thread::spawn(move || {

*res_body.lock().unwrap() = ResponseBody::Receiving(vec![]);
let mut new_body = vec![];
res.response.read_to_end(&mut new_body);

let mut body = res_body.lock().unwrap();
assert!(*body != ResponseBody::Empty);
*body = ResponseBody::Done(new_body);

// TODO: the vec storage format is much too slow for these operations,
// response.body needs to use something else before this code can be used
// *res_body.lock().unwrap() = ResponseBody::Receiving(vec![]);

// loop {
// match read_block(&mut res.response) {
// Ok(ReadResult::Payload(ref mut new_body)) => {
// if let ResponseBody::Receiving(ref mut body) = *res_body.lock().unwrap() {
// (body).append(new_body);
// }
// },
// Ok(ReadResult::EOF) | Err(_) => break
// }

// }

// let mut completed_body = res_body.lock().unwrap();
// if let ResponseBody::Receiving(ref body) = *completed_body {
// // TODO cloning seems sub-optimal, but I couldn't figure anything else out
// *res_body.lock().unwrap() = ResponseBody::Done((*body).clone());
// }
});
},
Err(e) =>
response.termination_reason = Some(TerminationReason::Fatal)
Expand Down
3 changes: 2 additions & 1 deletion components/net/fetch/response.rs
View file
Expand Up @@ -9,6 +9,7 @@ use std::ascii::AsciiExt;
use std::cell::{Cell, RefCell};
use std::rc::Rc;
use std::sync::mpsc::Receiver;
use std::sync::{Arc, Mutex};
use url::Url;

pub trait ResponseMethods {
Expand All @@ -24,7 +25,7 @@ impl ResponseMethods for Response {
url_list: RefCell::new(Vec::new()),
status: Some(StatusCode::Ok),
headers: Headers::new(),
body: RefCell::new(ResponseBody::Empty),
body: Arc::new(Mutex::new(ResponseBody::Empty)),
cache_state: CacheState::None,
https_state: HttpsState::None,
internal_response: None,
Expand Down
4 changes: 2 additions & 2 deletions components/net/http_loader.rs
View file
Expand Up @@ -101,12 +101,12 @@ pub fn factory(user_agent: String,
}
}

enum ReadResult {
pub enum ReadResult {
Payload(Vec<u8>),
EOF,
}

fn read_block<R: Read>(reader: &mut R) -> Result<ReadResult, ()> {
pub fn read_block<R: Read>(reader: &mut R) -> Result<ReadResult, ()> {
let mut buf = vec![0; 1024];

match reader.read(&mut buf) {
Expand Down
16 changes: 11 additions & 5 deletions components/net_traits/response.rs
View file
Expand Up @@ -8,6 +8,7 @@ use hyper::header::{AccessControlExposeHeaders, Headers};
use hyper::status::StatusCode;
use std::ascii::AsciiExt;
use std::cell::{Cell, RefCell};
use std::sync::{Arc, Mutex};
use url::Url;

/// [Response type](https://fetch.spec.whatwg.org/#concept-response-type)
Expand All @@ -31,7 +32,7 @@ pub enum TerminationReason {

/// The response body can still be pushed to after fetch
/// This provides a way to store unfinished response bodies
#[derive(Clone, Debug)]
#[derive(Clone, Debug, PartialEq)]
pub enum ResponseBody {
Empty, // XXXManishearth is this necessary, or is Done(vec![]) enough?
Receiving(Vec<u8>),
Expand Down Expand Up @@ -81,7 +82,7 @@ pub struct Response {
/// `None` can be considered a StatusCode of `0`.
pub status: Option<StatusCode>,
pub headers: Headers,
pub body: RefCell<ResponseBody>,
pub body: Arc<Mutex<ResponseBody>>,
pub cache_state: CacheState,
pub https_state: HttpsState,
/// [Internal response](https://fetch.spec.whatwg.org/#concept-internal-response), only used if the Response
Expand All @@ -100,7 +101,7 @@ impl Response {
url_list: RefCell::new(vec![]),
status: None,
headers: Headers::new(),
body: RefCell::new(ResponseBody::Empty),
body: Arc::new(Mutex::new(ResponseBody::Empty)),
cache_state: CacheState::None,
https_state: HttpsState::None,
internal_response: None,
Expand All @@ -115,6 +116,11 @@ impl Response {
}
}

pub fn wait_until_done(&self) {
while !self.body.lock().unwrap().is_done() && !self.is_network_error() {
}
}

pub fn get_actual_response(&self) -> &Response {
if self.return_internal.get() && self.internal_response.is_some() {
&**self.internal_response.as_ref().unwrap()
Expand Down Expand Up @@ -188,14 +194,14 @@ impl Response {
response.url = None;
response.headers = Headers::new();
response.status = None;
response.body = RefCell::new(ResponseBody::Empty);
response.body = Arc::new(Mutex::new(ResponseBody::Empty));
response.cache_state = CacheState::None;
},

ResponseType::OpaqueRedirect => {
response.headers = Headers::new();
response.status = None;
response.body = RefCell::new(ResponseBody::Empty);
response.body = Arc::new(Mutex::new(ResponseBody::Empty));
response.cache_state = CacheState::None;
}
}
Expand Down

0 comments on commit b187985

Please sign in to comment.