Replace servo-websocket by ws

This is heavily based on previous work done in #16012.

Fixes #14517

5 files changed, 155 insertions, 625 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 29043ccf16c..a67744d7376 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2268,7 +2268,6 @@ dependencies = [
  "profile_traits 0.0.1",
  "serde 1.0.66 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde_json 1.0.13 (registry+https://github.com/rust-lang/crates.io-index)",
- "servo-websocket 0.21.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "servo_allocator 0.0.1",
  "servo_arc 0.1.1",
  "servo_config 0.0.1",
@@ -2279,6 +2278,7 @@ dependencies = [
  "url 1.6.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "uuid 0.6.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "webrender_api 0.57.2 (git+https://github.com/servo/webrender)",
+ "ws 0.7.3 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -3268,21 +3268,6 @@ dependencies = [
 ]
 
 [[package]]
-name = "servo-websocket"
-version = "0.21.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-dependencies = [
- "base64 0.6.0 (registry+https://github.com/rust-lang/crates.io-index)",
- "bitflags 1.0.3 (registry+https://github.com/rust-lang/crates.io-index)",
- "byteorder 1.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
- "hyper 0.10.13 (registry+https://github.com/rust-lang/crates.io-index)",
- "rand 0.3.22 (registry+https://github.com/rust-lang/crates.io-index)",
- "sha1 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
- "unicase 1.4.0 (registry+https://github.com/rust-lang/crates.io-index)",
- "url 1.6.0 (registry+https://github.com/rust-lang/crates.io-index)",
-]
-
-[[package]]
 name = "servo_allocator"
 version = "0.0.1"
 dependencies = [
@@ -4152,6 +4137,7 @@ dependencies = [
  "httparse 1.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "log 0.3.9 (registry+https://github.com/rust-lang/crates.io-index)",
  "mio 0.6.12 (registry+https://github.com/rust-lang/crates.io-index)",
+ "openssl 0.9.24 (registry+https://github.com/rust-lang/crates.io-index)",
  "rand 0.3.22 (registry+https://github.com/rust-lang/crates.io-index)",
  "sha1 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "slab 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -4513,7 +4499,6 @@ dependencies = [
 "checksum servo-media-gstreamer 0.1.0 (git+https://github.com/servo/media)" = "<none>"
 "checksum servo-media-player 0.1.0 (git+https://github.com/servo/media)" = "<none>"
 "checksum servo-skia 0.30000019.0 (registry+https://github.com/rust-lang/crates.io-index)" = "00e9a17304c6181d04fdd76c2deecac41878cc929879a068711462b0e593b669"
-"checksum servo-websocket 0.21.0 (registry+https://github.com/rust-lang/crates.io-index)" = "6bac1e2295e72f0525147d993c626761811acf0441dac1cee8707f12dc7f3363"
 "checksum servo_media_derive 0.1.0 (git+https://github.com/servo/media)" = "<none>"
 "checksum sha1 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "cc30b1e1e8c40c121ca33b86c23308a090d19974ef001b4bf6e61fd1a0fb095c"
 "checksum shared_library 0.1.8 (registry+https://github.com/rust-lang/crates.io-index)" = "8254bf098ce4d8d7cc7cc6de438c5488adc5297e5b7ffef88816c0a91bd289c1"
diff --git a/components/net/Cargo.toml b/components/net/Cargo.toml
index 43c8b22c4c0..a2330c2aa62 100644
--- a/components/net/Cargo.toml
+++ b/components/net/Cargo.toml
@@ -41,13 +41,13 @@ servo_allocator = {path = "../allocator"}
 servo_arc = {path = "../servo_arc"}
 servo_config = {path = "../config"}
 servo_url = {path = "../url"}
-servo-websocket = { version = "0.21", default-features = false, features = ["sync"] }
 threadpool = "1.0"
 time = "0.1.17"
 unicase = "1.4.0"
 url = "1.2"
 uuid = {version = "0.6", features = ["v4"]}
 webrender_api = {git = "https://github.com/servo/webrender", features = ["ipc"]}
+ws = { version = "0.7", features = ["ssl"] }
 
 [dev-dependencies]
 embedder_traits = { path = "../embedder_traits", features = ["tests"] }
diff --git a/components/net/lib.rs b/components/net/lib.rs
index cdeb7d54595..928d7ae191a 100644
--- a/components/net/lib.rs
+++ b/components/net/lib.rs
@@ -40,7 +40,7 @@ extern crate unicase;
 extern crate url;
 extern crate uuid;
 extern crate webrender_api;
-extern crate websocket;
+extern crate ws;
 
 mod blob_loader;
 pub mod connector;
diff --git a/components/net/websocket_loader.rs b/components/net/websocket_loader.rs
index 77140631e7a..34e8f68d0b0 100644
--- a/components/net/websocket_loader.rs
+++ b/components/net/websocket_loader.rs
@@ -3,657 +3,205 @@
  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
 
 use cookie::Cookie;
-use fetch::methods::{should_be_blocked_due_to_bad_port, should_be_blocked_due_to_nosniff};
+use fetch::methods::should_be_blocked_due_to_bad_port;
 use hosts::replace_host;
-use http_loader::{HttpState, is_redirect_status, set_default_accept};
-use http_loader::{set_default_accept_language, set_request_cookies};
-use hyper::buffer::BufReader;
-use hyper::header::{CacheControl, CacheDirective, Connection, ConnectionOption};
-use hyper::header::{Headers, Host, SetCookie, Pragma, Protocol, ProtocolName, Upgrade};
-use hyper::http::h1::{LINE_ENDING, parse_response};
-use hyper::method::Method;
-use hyper::net::HttpStream;
-use hyper::status::StatusCode;
-use hyper::version::HttpVersion;
+use http_loader::HttpState;
+use hyper::header::{Headers, Host, SetCookie};
 use ipc_channel::ipc::{IpcReceiver, IpcSender};
-use net_traits::{CookieSource, MessageData, NetworkError};
+use net_traits::{CookieSource, MessageData};
 use net_traits::{WebSocketDomAction, WebSocketNetworkEvent};
-use net_traits::request::{Destination, RequestInit, RequestMode};
+use net_traits::request::{RequestInit, RequestMode};
 use servo_url::ServoUrl;
-use std::io::{self, Write};
-use std::net::TcpStream;
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::thread;
-use url::Position;
-use websocket::Message;
-use websocket::header::{Origin, WebSocketAccept, WebSocketKey, WebSocketProtocol, WebSocketVersion};
-use websocket::message::OwnedMessage;
-use websocket::receiver::{Reader as WsReader, Receiver as WsReceiver};
-use websocket::sender::{Sender as WsSender, Writer as WsWriter};
-use websocket::ws::dataframe::DataFrame;
-
-pub fn init(
-    req_init: RequestInit,
-    resource_event_sender: IpcSender<WebSocketNetworkEvent>,
-    dom_action_receiver: IpcReceiver<WebSocketDomAction>,
-    http_state: Arc<HttpState>
-) {
-    thread::Builder::new().name(format!("WebSocket connection to {}", req_init.url)).spawn(move || {
-        let channel = establish_a_websocket_connection(req_init, &http_state);
-        let (ws_sender, mut receiver) = match channel {
-            Ok((protocol_in_use, sender, receiver)) => {
-                let _ = resource_event_sender.send(WebSocketNetworkEvent::ConnectionEstablished { protocol_in_use });
-                (sender, receiver)
-            },
-            Err(e) => {
-                debug!("Failed to establish a WebSocket connection: {:?}", e);
-                let _ = resource_event_sender.send(WebSocketNetworkEvent::Fail);
-                return;
-            }
-
-        };
-
-        let initiated_close = Arc::new(AtomicBool::new(false));
-        let ws_sender = Arc::new(Mutex::new(ws_sender));
-
-        let initiated_close_incoming = initiated_close.clone();
-        let ws_sender_incoming = ws_sender.clone();
-        thread::spawn(move || {
-            for message in receiver.incoming_messages() {
-                let message = match message {
-                    Ok(m) => m,
-                    Err(e) => {
-                        debug!("Error receiving incoming WebSocket message: {:?}", e);
-                        let _ = resource_event_sender.send(WebSocketNetworkEvent::Fail);
-                        break;
-                    }
-                };
-                let message = match message {
-                    OwnedMessage::Text(_) => {
-                        MessageData::Text(String::from_utf8_lossy(&message.take_payload()).into_owned())
-                    },
-                    OwnedMessage::Binary(_) => MessageData::Binary(message.take_payload()),
-                    OwnedMessage::Ping(_) => {
-                        let pong = Message::pong(message.take_payload());
-                        ws_sender_incoming.lock().unwrap().send_message(&pong).unwrap();
-                        continue;
-                    },
-                    OwnedMessage::Pong(_) => continue,
-                    OwnedMessage::Close(ref msg) => {
-                        if !initiated_close_incoming.fetch_or(true, Ordering::SeqCst) {
-                            ws_sender_incoming.lock().unwrap().send_message(&message).unwrap();
-                        }
-                        let (code, reason) = match *msg {
-                            None => (None, "".into()),
-                            Some(ref data) => (Some(data.status_code), data.reason.clone())
-                        };
-                        let _ = resource_event_sender.send(WebSocketNetworkEvent::Close(code, reason));
-                        break;
-                    },
-                };
-                let _ = resource_event_sender.send(WebSocketNetworkEvent::MessageReceived(message));
-            }
-        });
-
-        while let Ok(dom_action) = dom_action_receiver.recv() {
-            match dom_action {
-                WebSocketDomAction::SendMessage(MessageData::Text(data)) => {
-                    ws_sender.lock().unwrap().send_message(&Message::text(data)).unwrap();
-                },
-                WebSocketDomAction::SendMessage(MessageData::Binary(data)) => {
-                    ws_sender.lock().unwrap().send_message(&Message::binary(data)).unwrap();
-                },
-                WebSocketDomAction::Close(code, reason) => {
-                    if !initiated_close.fetch_or(true, Ordering::SeqCst) {
-                        let message = match code {
-                            Some(code) => Message::close_because(code, reason.unwrap_or("".to_owned())),
-                            None => Message::close()
-                        };
-                        ws_sender.lock().unwrap().send_message(&message).unwrap();
-                    }
-                },
-            }
-        }
-    }).expect("Thread spawning failed");
+use url::Url;
+use ws::{CloseCode, Factory, Handler, Handshake, Message, Request, Response as WsResponse, Sender, WebSocket};
+use ws::{Error as WebSocketError, ErrorKind as WebSocketErrorKind, Result as WebSocketResult};
+
+/// A client for connecting to a websocket server
+#[derive(Clone)]
+struct Client<'a> {
+    origin: &'a str,
+    host: &'a Host,
+    protocols: &'a [String],
+    http_state: &'a Arc<HttpState>,
+    resource_url: &'a ServoUrl,
+    event_sender: &'a IpcSender<WebSocketNetworkEvent>,
+    protocol_in_use: Option<String>,
 }
 
-type Stream = HttpStream;
+impl<'a> Factory for Client<'a> {
+    type Handler = Self;
 
-// https://fetch.spec.whatwg.org/#concept-websocket-connection-obtain
-fn obtain_a_websocket_connection(url: &ServoUrl) -> Result<Stream, NetworkError> {
-    // Step 1.
-    let host = url.host_str().unwrap();
-
-    // Step 2.
-    let port = url.port_or_known_default().unwrap();
-
-    // Step 3.
-    // We did not replace the scheme by "http" or "https" in step 1 of
-    // establish_a_websocket_connection.
-    let secure = match url.scheme() {
-        "ws" => false,
-        "wss" => true,
-        _ => panic!("URL's scheme should be ws or wss"),
-    };
-
-    if secure {
-        return Err(NetworkError::Internal("WSS is disabled for now.".into()));
+    fn connection_made(&mut self, _: Sender) -> Self::Handler {
+        self.clone()
     }
 
-    // Steps 4-5.
-    let host = replace_host(host);
-    let tcp_stream = TcpStream::connect((&*host, port)).map_err(|e| {
-        NetworkError::Internal(format!("Could not connect to host: {}", e))
-    })?;
-    Ok(HttpStream(tcp_stream))
-}
-
-// https://fetch.spec.whatwg.org/#concept-websocket-establish
-fn establish_a_websocket_connection(
-    req_init: RequestInit,
-    http_state: &HttpState
-) -> Result<(Option<String>, WsWriter<HttpStream>, WsReader<HttpStream>), NetworkError>
-{
-    let protocols = match req_init.mode {
-        RequestMode::WebSocket { protocols } => protocols.clone(),
-        _ => panic!("Received a RequestInit with a non-websocket mode in websocket_loader"),
-    };
-    // Steps 1 is not really applicable here, given we don't exactly go
-    // through the same infrastructure as the Fetch spec.
-
-    // Step 2, slimmed down because we don't go through the whole Fetch infra.
-    let mut headers = Headers::new();
-
-    // Step 3.
-    headers.set(Upgrade(vec![Protocol::new(ProtocolName::WebSocket, None)]));
-
-    // Step 4.
-    headers.set(Connection(vec![ConnectionOption::ConnectionHeader("upgrade".into())]));
-
-    // Step 5.
-    let key_value = WebSocketKey::new();
-
-    // Step 6.
-    headers.set(key_value);
-
-    // Step 7.
-    headers.set(WebSocketVersion::WebSocket13);
-
-    // Step 8.
-    if !protocols.is_empty() {
-        headers.set(WebSocketProtocol(protocols.clone()));
+    fn connection_lost(&mut self, _: Self::Handler) {
+        let _ = self.event_sender.send(WebSocketNetworkEvent::Fail);
     }
+}
 
-    // Steps 9-10.
-    // TODO: handle permessage-deflate extension.
 
-    // Step 11 and network error check from step 12.
-    let response = fetch(req_init.url, req_init.origin.ascii_serialization(), headers, http_state)?;
+impl<'a> Handler for Client<'a> {
+    fn build_request(&mut self, url: &Url) -> WebSocketResult<Request> {
+        let mut req = Request::from_url(url)?;
+        req.headers_mut().push(("Origin".to_string(), self.origin.as_bytes().to_owned()));
+        req.headers_mut().push(("Host".to_string(), format!("{}", self.host).as_bytes().to_owned()));
 
-    // Step 12, the status code check.
-    if response.status != StatusCode::SwitchingProtocols {
-        return Err(NetworkError::Internal("Response's status should be 101.".into()));
-    }
+        for protocol in self.protocols {
+            req.add_protocol(protocol);
+        };
 
-    // Step 13.
-    if !protocols.is_empty() {
-        if response.headers.get::<WebSocketProtocol>().map_or(true, |protocols| protocols.is_empty()) {
-            return Err(NetworkError::Internal(
-                "Response's Sec-WebSocket-Protocol header is missing, malformed or empty.".into()));
+        let mut cookie_jar = self.http_state.cookie_jar.write().unwrap();
+        if let Some(cookie_list) = cookie_jar.cookies_for_url(self.resource_url, CookieSource::HTTP) {
+            req.headers_mut().push(("Cookie".into(), cookie_list.as_bytes().to_owned()))
         }
-    }
 
-    // Step 14.2.
-    let upgrade_header = response.headers.get::<Upgrade>().ok_or_else(|| {
-        NetworkError::Internal("Response should have an Upgrade header.".into())
-    })?;
-    if upgrade_header.len() != 1 {
-        return Err(NetworkError::Internal("Response's Upgrade header should have only one value.".into()));
-    }
-    if upgrade_header[0].name != ProtocolName::WebSocket {
-        return Err(NetworkError::Internal("Response's Upgrade header value should be \"websocket\".".into()));
+        Ok(req)
     }
 
-    // Step 14.3.
-    let connection_header = response.headers.get::<Connection>().ok_or_else(|| {
-        NetworkError::Internal("Response should have a Connection header.".into())
-    })?;
-    let connection_includes_upgrade = connection_header.iter().any(|option| {
-        match *option {
-            ConnectionOption::ConnectionHeader(ref option) => *option == "upgrade",
-            _ => false,
+    fn on_open(&mut self, shake: Handshake) -> WebSocketResult<()> {
+        let mut headers = Headers::new();
+        for &(ref name, ref value) in shake.response.headers().iter() {
+            headers.set_raw(name.clone(), vec![value.clone()]);
         }
-    });
-    if !connection_includes_upgrade {
-        return Err(NetworkError::Internal("Response's Connection header value should include \"upgrade\".".into()));
-    }
 
-    // Step 14.4.
-    let accept_header = response.headers.get::<WebSocketAccept>().ok_or_else(|| {
-        NetworkError::Internal("Response should have a Sec-Websocket-Accept header.".into())
-    })?;
-    if *accept_header != WebSocketAccept::new(&key_value) {
-        return Err(NetworkError::Internal(
-            "Response's Sec-WebSocket-Accept header value did not match the sent key.".into()));
-    }
-
-    // Step 14.5.
-    // TODO: handle permessage-deflate extension.
-    // We don't support any extension, so we fail at the mere presence of
-    // a Sec-WebSocket-Extensions header.
-    if response.headers.get_raw("Sec-WebSocket-Extensions").is_some() {
-        return Err(NetworkError::Internal(
-            "Response's Sec-WebSocket-Extensions header value included unsupported extensions.".into()));
-    }
-
-    // Step 14.6.
-    let protocol_in_use = if let Some(response_protocols) = response.headers.get::<WebSocketProtocol>() {
-        for replied in &**response_protocols {
-            if !protocols.iter().any(|requested| requested.eq_ignore_ascii_case(replied)) {
-                return Err(NetworkError::Internal(
-                    "Response's Sec-WebSocket-Protocols contain values that were not requested.".into()));
+         if let Some(cookies) = headers.get::<SetCookie>() {
+            let mut jar = self.http_state.cookie_jar.write().unwrap();
+            for cookie in &**cookies {
+                if let Some(cookie) =
+                    Cookie::from_cookie_string(cookie.clone(), self.resource_url, CookieSource::HTTP)
+                {
+                    jar.push(cookie, self.resource_url, CookieSource::HTTP);
+                }
             }
         }
-        response_protocols.first().cloned()
-    } else {
-        None
-    };
-
-    let sender = WsSender::new(true);
-    let writer = WsWriter {
-        stream: response.writer,
-        sender
-    };
-
-    let receiver = WsReceiver::new(false);
-    let reader = WsReader {
-        stream: response.reader,
-        receiver,
-    };
-
-    Ok((protocol_in_use, writer, reader))
-}
-
-struct Response {
-    status: StatusCode,
-    headers: Headers,
-    reader: BufReader<Stream>,
-    writer: Stream,
-}
-
-// https://fetch.spec.whatwg.org/#concept-fetch
-fn fetch(url: ServoUrl,
-         origin: String,
-         mut headers: Headers,
-         http_state: &HttpState)
-         -> Result<Response, NetworkError> {
-    // Step 1.
-    // TODO: handle request's window.
-
-    // Step 2.
-    // TODO: handle request's origin.
-
-    // Step 3.
-    set_default_accept(Destination::None, &mut headers);
-
-    // Step 4.
-    set_default_accept_language(&mut headers);
-
-    // Step 5.
-    // TODO: handle request's priority.
 
-    // Step 6.
-    // Not applicable: not a navigation request.
-
-    // Step 7.
-    // We know this is a subresource request.
-    {
-        // Step 7.1.
-        // Not applicable: client hints list is empty.
-
-        // Steps 7.2-3.
-        // TODO: handle fetch groups.
-    }
-
-    // Step 8.
-    main_fetch(url, origin, headers, http_state)
-}
-
-// https://fetch.spec.whatwg.org/#concept-main-fetch
-fn main_fetch(url: ServoUrl,
-              origin: String,
-              mut headers: Headers,
-              http_state: &HttpState)
-              -> Result<Response, NetworkError> {
-    // Step 1.
-    let mut response = None;
-
-    // Step 2.
-    // Not applicable: request’s local-URLs-only flag is unset.
-
-    // Step 3.
-    // TODO: handle content security policy violations.
-
-    // Step 4.
-    // TODO: handle upgrade to a potentially secure URL.
-
-    // Step 5.
-    if should_be_blocked_due_to_bad_port(&url) {
-        response = Some(Err(NetworkError::Internal("Request should be blocked due to bad port.".into())));
+        let _ = self.event_sender.send(
+            WebSocketNetworkEvent::ConnectionEstablished { protocol_in_use: self.protocol_in_use.clone() });
+        Ok(())
     }
-    // TODO: handle blocking as mixed content.
-    // TODO: handle blocking by content security policy.
-
-    // Steps 6-8.
-    // TODO: handle request's referrer policy.
-
-    // Step 9.
-    // Not applicable: request's current URL's scheme is not "ftp".
-
-    // Step 10.
-    // TODO: handle known HSTS host domain.
-
-    // Step 11.
-    // Not applicable: request's synchronous flag is set.
-
-    // Step 12.
-    let mut response = response.unwrap_or_else(|| {
-        // We must run the first sequence of substeps, given request's mode
-        // is "websocket".
-
-        // Step 12.1.
-        // Not applicable: the response is never exposed to the Web so it
-        // doesn't need to be filtered at all.
-
-        // Step 12.2.
-        scheme_fetch(&url, origin, &mut headers, http_state)
-    });
-
-    // Step 13.
-    // Not applicable: recursive flag is unset.
-
-    // Step 14.
-    // Not applicable: the response is never exposed to the Web so it doesn't
-    // need to be filtered at all.
 
-    // Steps 15-16.
-    // Not applicable: no need to maintain an internal response.
+     fn on_message(&mut self, message: Message) -> WebSocketResult<()> {
+        let message = match message {
+            Message::Text(message) => MessageData::Text(message),
+            Message::Binary(message) => MessageData::Binary(message),
+        };
+        let _ = self.event_sender.send(WebSocketNetworkEvent::MessageReceived(message));
 
-    // Step 17.
-    if response.is_ok() {
-        // TODO: handle blocking as mixed content.
-        // TODO: handle blocking by content security policy.
-        // Not applicable: blocking due to MIME type matters only for scripts.
-        if should_be_blocked_due_to_nosniff(Destination::None, &headers) {
-            response = Err(NetworkError::Internal("Request should be blocked due to nosniff.".into()));
-        }
+        Ok(())
     }
 
-    // Step 18.
-    // Not applicable: we don't care about the body at all.
-
-    // Step 19.
-    // Not applicable: request's integrity metadata is the empty string.
-
-    // Step 20.
-    // TODO: wait for response's body here, maybe?
-    response
-}
-
-// https://fetch.spec.whatwg.org/#concept-scheme-fetch
-fn scheme_fetch(url: &ServoUrl,
-               origin: String,
-               headers: &mut Headers,
-               http_state: &HttpState)
-               -> Result<Response, NetworkError> {
-    // In the case of a WebSocket request, HTTP fetch is always used.
-    http_fetch(url, origin, headers, http_state)
-}
-
-// https://fetch.spec.whatwg.org/#concept-http-fetch
-fn http_fetch(url: &ServoUrl,
-              origin: String,
-              headers: &mut Headers,
-              http_state: &HttpState)
-              -> Result<Response, NetworkError> {
-    // Step 1.
-    // Not applicable: with step 3 being useless here, this one is too.
-
-    // Step 2.
-    // Not applicable: we don't need to maintain an internal response.
-
-    // Step 3.
-    // Not applicable: request's service-workers mode is "none".
 
-    // Step 4.
-    // There cannot be a response yet at this point.
-    let mut response = {
-        // Step 4.1.
-        // Not applicable: CORS-preflight flag is unset.
-
-        // Step 4.2.
-        // Not applicable: request's redirect mode is "error".
-
-        // Step 4.3.
-        let response = http_network_or_cache_fetch(url, origin, headers, http_state);
-
-        // Step 4.4.
-        // Not applicable: CORS flag is unset.
-
-        response
-    };
-
-    // Step 5.
-    if response.as_ref().ok().map_or(false, |response| is_redirect_status(response.status)) {
-        // Step 5.1.
-        // Not applicable: the connection does not use HTTP/2.
-
-        // Steps 5.2-4.
-        // Not applicable: matters only if request's redirect mode is not "error".
-
-        // Step 5.5.
-        // Request's redirect mode is "error".
-        response = Err(NetworkError::Internal("Response should not be a redirection.".into()));
+    fn on_error(&mut self, err: WebSocketError) {
+        debug!("Error in WebSocket communication: {:?}", err);
+        let _ = self.event_sender.send(WebSocketNetworkEvent::Fail);
     }
 
-    // Step 6.
-    response
-}
-
-// https://fetch.spec.whatwg.org/#concept-http-network-or-cache-fetch
-fn http_network_or_cache_fetch(url: &ServoUrl,
-                               origin: String,
-                               headers: &mut Headers,
-                               http_state: &HttpState)
-                               -> Result<Response, NetworkError> {
-    // Steps 1-3.
-    // Not applicable: we don't even have a request yet, and there is no body
-    // in a WebSocket request.
-
-    // Step 4.
-    // Not applicable: credentials flag is always set
-    // because credentials mode is "include."
 
-    // Steps 5-9.
-    // Not applicable: there is no body in a WebSocket request.
-
-    // Step 10.
-    // TODO: handle header Referer.
-
-    // Step 11.
-    // Request's mode is "websocket".
-    headers.set(Origin(origin));
-
-    // Step 12.
-    // TODO: handle header User-Agent.
-
-    // Steps 13-14.
-    // Not applicable: request's cache mode is "no-store".
-
-    // Step 15.
-    {
-        // Step 15.1.
-        // We know there is no Pragma header yet.
-        headers.set(Pragma::NoCache);
-
-        // Step 15.2.
-        // We know there is no Cache-Control header yet.
-        headers.set(CacheControl(vec![CacheDirective::NoCache]));
+    fn on_response(&mut self, res: &WsResponse) -> WebSocketResult<()> {
+        let protocol_in_use = res.protocol()?;
+        if let Some(protocol_name) = protocol_in_use {
+            let protocol_name = protocol_name.to_lowercase();
+            if !self.protocols.is_empty() && !self.protocols.iter().any(|p| protocol_name == (*p).to_lowercase()) {
+                let error = WebSocketError::new(WebSocketErrorKind::Protocol,
+                                                "Protocol in Use not in client-supplied protocol list");
+                return Err(error);
+            }
+            self.protocol_in_use = Some(protocol_name);
+        }
+        Ok(())
     }
 
-    // Step 16.
-    // TODO: handle Accept-Encoding.
-    // Not applicable: Connection header is already present.
-    // TODO: handle DNT.
-    headers.set(Host {
-        hostname: url.host_str().unwrap().to_owned(),
-        port: url.port(),
-    });
-
-    // Step 17.
-    // Credentials flag is set.
-    {
-        // Step 17.1.
-        // TODO: handle user agent configured to block cookies.
-        set_request_cookies(&url, headers, &http_state.cookie_jar);
-
-        // Steps 17.2-6.
-        // Not applicable: request has no Authorization header.
+    fn on_close(&mut self, code: CloseCode, reason: &str) {
+        debug!("Connection closing due to ({:?}) {}", code, reason);
+        let _ = self.event_sender.send(WebSocketNetworkEvent::Close(Some(code.into()), reason.to_owned()));
     }
-
-    // Step 18.
-    // TODO: proxy-authentication entry.
-
-    // Step 19.
-    // Not applicable: with step 21 being useless, this one is too.
-
-    // Step 20.
-    // Not applicable: revalidatingFlag is only useful if step 21 is.
-
-    // Step 21.
-    // Not applicable: cache mode is "no-store".
-
-    // Step 22.
-    // There is no response yet.
-    let response = {
-        // Step 22.1.
-        // Not applicable: cache mode is "no-store".
-
-        // Step 22.2.
-        let forward_response = http_network_fetch(url, headers, http_state);
-
-        // Step 22.3.
-        // Not applicable: request's method is not unsafe.
-
-        // Step 22.4.
-        // Not applicable: revalidatingFlag is unset.
-
-        // Step 22.5.
-        // There is no response yet and the response should not be cached.
-        forward_response
-    };
-
-    // Step 23.
-    // TODO: handle 401 status when request's window is not "no-window".
-
-    // Step 24.
-    // TODO: handle 407 status when request's window is not "no-window".
-
-    // Step 25.
-    // Not applicable: authentication-fetch flag is unset.
-
-    // Step 26.
-    response
 }
 
-// https://fetch.spec.whatwg.org/#concept-http-network-fetch
-fn http_network_fetch(url: &ServoUrl,
-                      headers: &Headers,
-                      http_state: &HttpState)
-                      -> Result<Response, NetworkError> {
-    // Step 1.
-    // Not applicable: credentials flag is set.
-
-    // Steps 2-3.
-    // Request's mode is "websocket".
-    let connection = obtain_a_websocket_connection(url)?;
-
-    // Step 4.
-    // Not applicable: request’s body is null.
-
-    // Step 5.
-    let response = make_request(connection, url, headers)?;
-
-    // Steps 6-12.
-    // Not applicable: correct WebSocket responses don't have a body.
-
-    // Step 13.
-    // TODO: handle response's CSP list.
-
-    // Step 14.
-    // Not applicable: request's cache mode is "no-store".
+pub fn init(
+    req_init: RequestInit,
+    resource_event_sender: IpcSender<WebSocketNetworkEvent>,
+    dom_action_receiver: IpcReceiver<WebSocketDomAction>,
+    http_state: Arc<HttpState>
+) {
+    thread::Builder::new().name(format!("WebSocket connection to {}", req_init.url)).spawn(move || {
+        let protocols = match req_init.mode {
+            RequestMode::WebSocket { protocols } => protocols.clone(),
+            _ => panic!("Received a RequestInit with a non-websocket mode in websocket_loader"),
+        };
 
-    // Step 15.
-    if let Some(cookies) = response.headers.get::<SetCookie>() {
-        let mut jar = http_state.cookie_jar.write().unwrap();
-        for cookie in &**cookies {
-            if let Some(cookie) = Cookie::from_cookie_string(cookie.clone(), url, CookieSource::HTTP) {
-                jar.push(cookie, url, CookieSource::HTTP);
-            }
+        let scheme = req_init.url.scheme();
+        let mut req_url = req_init.url.clone();
+        if scheme == "ws" {
+            req_url.as_mut_url().set_scheme("http").unwrap();
+        } else if scheme == "wss" {
+            req_url.as_mut_url().set_scheme("https").unwrap();
         }
-    }
 
-    // Step 16.
-    // Not applicable: correct WebSocket responses don't have a body.
+        if should_be_blocked_due_to_bad_port(&req_url) {
+            debug!("Failed to establish a WebSocket connection: port blocked");
+            let _ = resource_event_sender.send(WebSocketNetworkEvent::Fail);
+            return;
+        }
 
-    // Step 17.
-    Ok(response)
-}
+        let host = replace_host(req_init.url.host_str().unwrap());
+        let mut net_url = req_init.url.clone().into_url();
+        net_url.set_host(Some(&host)).unwrap();
 
-fn make_request(mut stream: Stream,
-                url: &ServoUrl,
-                headers: &Headers)
-                -> Result<Response, NetworkError> {
-    write_request(&mut stream, url, headers).map_err(|e| {
-        NetworkError::Internal(format!("Request could not be sent: {}", e))
-    })?;
+        let host = Host {
+            hostname: req_init.url.host_str().unwrap().to_owned(),
+            port: req_init.url.port_or_known_default(),
+        };
 
-    // FIXME: Stream isn't supposed to be cloned.
-    let writer = stream.clone();
+        let client = Client {
+            origin: &req_init.origin.ascii_serialization(),
+            host: &host,
+            protocols: &protocols,
+            http_state: &http_state,
+            resource_url: &req_init.url,
+            event_sender: &resource_event_sender,
+            protocol_in_use: None,
+        };
+        let mut ws = WebSocket::new(client).unwrap();
 
-    // FIXME: BufReader from hyper isn't supposed to be used.
-    let mut reader = BufReader::new(stream);
+        if let Err(e) = ws.connect(net_url) {
+            debug!("Failed to establish a WebSocket connection: {:?}", e);
+            return;
+        };
 
-    let head = parse_response(&mut reader).map_err(|e| {
-        NetworkError::Internal(format!("Response could not be read: {}", e))
-    })?;
+        let ws_sender = ws.broadcaster();
+        let initiated_close = Arc::new(AtomicBool::new(false));
 
-    // This isn't in the spec, but this is the correct thing to do for WebSocket requests.
-    if head.version != HttpVersion::Http11 {
-        return Err(NetworkError::Internal("Response's HTTP version should be HTTP/1.1.".into()));
-    }
+        thread::spawn(move || {
+            while let Ok(dom_action) = dom_action_receiver.recv() {
+                match dom_action {
+                    WebSocketDomAction::SendMessage(MessageData::Text(data)) => {
+                        ws_sender.send(Message::text(data)).unwrap();
+                    },
+                    WebSocketDomAction::SendMessage(MessageData::Binary(data)) => {
+                        ws_sender.send(Message::binary(data)).unwrap();
+                    },
+                    WebSocketDomAction::Close(code, reason) => {
+                        if !initiated_close.fetch_or(true, Ordering::SeqCst) {
+                            match code {
+                                Some(code) => {
+                                    ws_sender.close_with_reason(code.into(), reason.unwrap_or("".to_owned())).unwrap()
+                                },
+                                None => ws_sender.close(CloseCode::Status).unwrap(),
+                            };
+                        }
+                    },
+                }
+            }
+        });
 
-    // FIXME: StatusCode::from_u16 isn't supposed to be used.
-    let status = StatusCode::from_u16(head.subject.0);
-    Ok(Response {
-        status: status,
-        headers: head.headers,
-        reader: reader,
-        writer: writer,
-    })
+        if let Err(e) = ws.run() {
+            debug!("Failed to run WebSocket: {:?}", e);
+            let _ = resource_event_sender.send(WebSocketNetworkEvent::Fail);
+        };
+    }).expect("Thread spawning failed");
 }
 
-fn write_request(stream: &mut Stream,
-                 url: &ServoUrl,
-                 headers: &Headers)
-                 -> io::Result<()> {
-    // Write "GET /foo/bar HTTP/1.1\r\n".
-    let method = Method::Get;
-    let request_uri = &url.as_url()[Position::BeforePath..Position::AfterQuery];
-    let version = HttpVersion::Http11;
-    write!(stream, "{} {} {}{}", method, request_uri, version, LINE_ENDING)?;
-
-    // Write the headers.
-    write!(stream, "{}{}", headers, LINE_ENDING)
-}
diff --git a/tests/wpt/metadata/websockets/interfaces/WebSocket/close/close-connecting.html.ini b/tests/wpt/metadata/websockets/interfaces/WebSocket/close/close-connecting.html.ini
index 1c04bf3872a..e1e790fd16c 100644
--- a/tests/wpt/metadata/websockets/interfaces/WebSocket/close/close-connecting.html.ini
+++ b/tests/wpt/metadata/websockets/interfaces/WebSocket/close/close-connecting.html.ini
@@ -1,8 +1,5 @@
 [close-connecting.html]
   type: testharness
-  [WebSockets: close() when connecting]
-    expected: FAIL
-
 
 [close-connecting.html?wss]
   type: testharness