http_client.rc

// Copyright 2013 The Servo Project Developers. See the COPYRIGHT
// file at the top-level directory of this distribution.
//
// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
// option. This file may not be copied, modified, or distributed
// except according to those terms.

/*!

A simple HTTP client in Rust

*/
#[link(name = "http_client",
       vers = "0.1",
       uuid = "3bfdcc24-ca55-447b-a9e1-8f7ac37c222e")];

#[crate_type = "lib"];

#[allow(non_uppercase_statics)];

extern mod extra;

use std::ptr::to_unsafe_ptr;
use std::result;
use std::result::{Result, Ok, Err};
use std::cell::Cell;
use std::str;
use extra::net::ip::{
    get_addr, format_addr,
    IpAddr, IpGetAddrErr, Ipv4, Ipv6
};
use extra::net::tcp::{TcpErrData, TcpSocket};
use extra::net::url::Url;
use extra::uv_global_loop;
use connection::{
    Connection, ConnectionFactory, UvConnectionFactory
};
use parser::{Parser, ParserCallbacks};
use request::build_request;

pub mod connection;
pub mod parser;
pub mod response_headers;
pub mod http_parser;
pub mod request;

pub static timeout: uint = 2000;

/// HTTP status codes
#[deriving(Eq)]
pub enum StatusCode {
    StatusContinue = 100,
    StatusSwitchingProtocols = 101,
    StatusOk = 200,
    StatusCreated = 201,
    StatusAccepted = 202,
    StatusNonAuthoritative = 203,
    StatusNoContent = 204,
    StatusResetContent = 205,
    StatusPartialContent = 206,
    StatusMultipleChoices = 300,
    StatusMovedPermanently = 301,
    StatusFound = 302,
    StatusSeeOther = 303,
    StatusNotModified = 304,
    StatusUseProxy = 305,
    StatusUnused = 306,
    StatusTemporaryRedirect = 307,
    StatusBadRequest = 400,
    StatusUnauthorized = 401,
    StatusPaymentRequired = 402,
    StatusForbidden = 403,
    StatusNotFound = 404,
    StatusNotAcceptable = 405,
    StatusProxyAuthenticationRequired = 407,
    StatusRequestTimeout = 408,
    StatusConflict = 409,
    StatusGone = 410,
    StatusLengthRequired = 411,
    StatusPreconditionFailed = 412,
    StatusRequestEntityTooLarge = 413,
    StatusRequestURITooLong = 414,
    StatusUnsupportedMediaType = 415,
    StatusRequestedRangeNotSatisfiable = 416,
    StatusExpectationFailed = 417,
    StatusInternalServerError = 500,
    StatusNotImplemented = 501,
    StatusBadGateway = 502,
    StatusServiceUnavailable = 503,
    StatusGatewayTimeout = 504,
    StatusHTTPVersionNotSupported = 505,
    StatusUnknown
}

/// HTTP request error conditions
#[deriving(Eq)]
pub enum RequestError {
    ErrorDnsResolution,
    ErrorConnect,
    ErrorMisc
}

/// Request
#[deriving(Eq)]
pub enum RequestEvent {
    Status(StatusCode),
    Payload(Cell<~[u8]>),
    Error(RequestError)
}

pub type DnsResolver = @fn(host: ~str) -> Result<~[IpAddr], IpGetAddrErr>;

pub fn uv_dns_resolver() -> DnsResolver {
    let r: DnsResolver = |host: ~str| {
        let iotask = uv_global_loop::get();
        get_addr(host.to_str(), &iotask)
    };
    return r;
}

pub fn uv_http_request(url: Url) -> HttpRequest<TcpSocket, UvConnectionFactory> {
    HttpRequest(uv_dns_resolver(), UvConnectionFactory, url)
}

#[allow(non_implicitly_copyable_typarams)]
pub struct HttpRequest<C, CF> {
    resolve_ip_addr: DnsResolver,
    connection_factory: CF,
    url: Url,
    parser: Parser,
    cb: @fn(ev: RequestEvent)
}

pub fn HttpRequest<C: Connection, CF: ConnectionFactory<C>>(resolver: DnsResolver,
                                                            connection_factory: CF,
                                                            url: Url) ->
                                                            HttpRequest<C,CF> {
    HttpRequest {
        resolve_ip_addr: resolver,
        connection_factory: connection_factory,
        url: url,
        parser: Parser(),
        cb: |_event| { }
    }
}

#[allow(non_implicitly_copyable_typarams)]
impl<C: Connection, CF: ConnectionFactory<C>> HttpRequest<C, CF> {
    pub fn begin(&mut self, cb: @fn(ev: RequestEvent)) {
        debug!("http_client: looking up url %?", self.url.to_str());
        let ip_addr = match self.get_ip() {
          Ok(addr) => { copy addr }
          Err(e) => { cb(Error(e)); return }
        };

        debug!("http_client: using IP %? for %?", format_addr(&ip_addr), self.url.to_str());

        let socket = {
            debug!("http_client: connecting to %?", ip_addr);
            let socket = self.connection_factory.connect(copy ip_addr, 80);
            if socket.is_ok() {
                result::unwrap(socket)
            } else {
                debug!("http_client: unable to connect to %?: %?", ip_addr, socket);
                cb(Error(ErrorConnect));
                return;
            }
        };

        debug!("http_client: got socket for %?", ip_addr);

        let request_header = build_request(copy self.url);
        debug!("http_client: writing request header: %?", request_header);
        let request_header_bytes = request_header.as_bytes().to_owned();
        match socket.write_(request_header_bytes) {
          result::Ok(*) => { }
          result::Err(*) => {
            // FIXME: Need test
            cb(Error(ErrorMisc));
            return;
          }
        }

        let read_port = {
            let read_port = socket.read_start_();
            if read_port.is_ok() {
                result::unwrap(read_port)
            } else {
                cb(Error(ErrorMisc));
                return;
            }
        };

        // This unsafety is unfortunate but we can't capture self
        // into shared closures
        let unsafe_self = to_unsafe_ptr(&self);
        let callbacks = ParserCallbacks {
            on_message_begin: || unsafe { (*unsafe_self).on_message_begin() },
            on_url: |data| unsafe { (*unsafe_self).on_url(data) },
            on_header_field: |data| unsafe { (*unsafe_self).on_header_field(data) },
            on_header_value: |data| unsafe { (*unsafe_self).on_header_value(data) },
            on_headers_complete: || unsafe { (*unsafe_self).on_headers_complete() },
            on_body: |data| unsafe { (*unsafe_self).on_body(data) },
            on_message_complete: || unsafe { (*unsafe_self).on_message_complete() }
        };

        // Set the callback used by the parser event handlers
        self.cb = cb;

        loop {
            let next_data = read_port.recv();

            if next_data.is_ok() {
                let next_data = result::unwrap(next_data);
                debug!("data: %?", str::from_bytes(next_data));
                let bytes_parsed = self.parser.execute(next_data, &callbacks);
                if bytes_parsed != next_data.len() {
                    // FIXME: Need tests
                    fail!(~"http parse failure");
                }
            } else {
                debug!("http_client: read error: %?", next_data);

                // This method of detecting EOF is lame
                match next_data {
                  result::Err(TcpErrData {err_name: ~"EOF", _}) => {
                    self.parser.execute([], &callbacks);
                    break;
                  }
                  _ => {
                    // FIXME: Need tests and error handling
                    socket.read_stop_(read_port);
                    cb(Error(ErrorMisc));
                    return;
                  }
                }
            }
        }
        socket.read_stop_(read_port);
    }

    pub fn get_ip(&self) -> Result<IpAddr, RequestError> {
        let ip_addrs = (self.resolve_ip_addr)(copy self.url.host);
        if ip_addrs.is_ok() {
            let ip_addrs = result::unwrap(ip_addrs);
            // FIXME: This log crashes
            //#debug("http_client: got IP addresses for %?: %?", self.url, ip_addrs);
            if ip_addrs.len() != 0 {
                // FIXME: Which address should we really pick?
                let best_ip = do ip_addrs.iter().find_ |ip| {
                    match **ip {
                      Ipv4(*) => { true }
                      Ipv6(*) => { false }
                    }
                };

                if best_ip.is_some() {
                    return Ok(*best_ip.unwrap());
                } else {
                    // FIXME: Need test
                    return Err(ErrorMisc);
                }
            } else {
                debug!("http_client: got no IP addresses for %?", self.url);
                // FIXME: Need test
                return Err(ErrorMisc);
            }
        } else {
            debug!("http_client: DNS lookup failure: %?", ip_addrs.get_err());
            return Err(ErrorDnsResolution);
        }
    }

    pub fn on_message_begin(&self) -> bool {
        debug!("on_message_begin");
        true
    }

    pub fn on_url(&self, _data: ~[u8]) -> bool {
        debug!("on_url");
        true
    }

    pub fn on_header_field(&self, data: ~[u8]) -> bool {
        let header_field = str::from_bytes(data);
        debug!("on_header_field: %?", header_field);
        true
    }

    pub fn on_header_value(&self, data: ~[u8]) -> bool {
        let header_value = str::from_bytes(data);
        debug!("on_header_value: %?", header_value);
        true
    }

    pub fn on_headers_complete(&self) -> bool {
        debug!("on_headers_complete");
        true
    }

    pub fn on_body(&self, data: ~[u8]) -> bool {
        debug!("on_body");
        let the_payload = Payload(Cell::new(data));
        (self.cb)(the_payload);
        true
    }

    pub fn on_message_complete(&self) -> bool {
        debug!("on_message_complete");
        let status_code = self.parser.status_code();

        let status = match status_code {
          200 => { StatusOk },
          302 => { StatusFound },
          _ => { StatusUnknown }
        };
        (self.cb)(Status(status));
        true
    }
}

#[allow(non_implicitly_copyable_typarams)]
pub fn sequence<C: Connection, CF: ConnectionFactory<C>>(request: &mut HttpRequest<C, CF>) ->
    ~[RequestEvent] {

    let events = @mut ~[];
    do request.begin |event| {
        events.push(event)
    }
    return copy *events;
}

#[test]
#[allow(non_implicitly_copyable_typarams)]
pub fn test_resolve_error() {
    use extra::net::url;
    let url = url::from_str("http://example.com_not_real/").get();
    let mut request = uv_http_request(url);
    let events = sequence(&mut request);

    assert!(events == ~[
        Error(ErrorDnsResolution),
    ]);
}

#[test]
#[allow(non_implicitly_copyable_typarams)]
pub fn test_connect_error() {
    // This address is invalid because the first octet
    // of a class A address cannot be 0
    use extra::net::url;
    let url = url::from_str("http://0.42.42.42/").get();
    let mut request = uv_http_request(url);
    let events = sequence(&mut request);

    assert!(events == ~[
        Error(ErrorConnect),
    ]);
}

#[test]
#[ignore(reason = "no external internet on build slaves")]
#[allow(non_implicitly_copyable_typarams)]
pub fn test_connect_success() {
    use extra::net::url;
    let url = url::from_str("http://www.google.com/").get();
    let mut request = uv_http_request(url);
    let events = sequence(&mut request);

    for events.iter().advance |ev| {
        match *ev {
          Error(*) => { fail!() },
          Status(status) => match status {
              StatusOk => { },
              StatusFound => { fail!(~"status found, expected OK (200)") },
              StatusUnknown => { fail!(~"status unknown, expected OK (200)") }
              _ => { fail!(~"Unexpected HTTP Status code returned.")}
          },
          _ => { }
        }
    }
}

#[test]
#[ignore(reason = "ICE")]
#[allow(non_implicitly_copyable_typarams)]
pub fn test_simple_response() {
    use extra::net::url;
    use connection::{MockConnection, MockConnectionFactory};
    use std::comm;
    let _url = url::from_str("http://whatever/").get();
    let _mock_connection = MockConnection {
        write_fn: |_data| { Ok(()) },
        read_start_fn: || {
            let (port, chan) = comm::stream();

            let response = ~"HTTP/1.0 200 OK\
                            \
                            Test";
            chan.send(Ok(response.as_bytes().to_owned()));

            Ok(@port)
        },
        read_stop_fn: |_port| { Ok(()) }
    };

    let _mock_connection_factory = MockConnectionFactory {
        connect_fn: |_ip, _port| {

            // FIXME this doesn't work
            fail!();//ok(mock_connection)
        }
    };
}