Add tokio support.

.gitignore

@@ -1,6 +1,9 @@
 /target
 /Cargo.lock
 
+# emacs
+*.#*.rs
+
 # Windows image file caches
 Thumbs.db
 ehthumbs.db

.travis.yml

@@ -7,7 +7,7 @@ before_script:
 
 script:
     - cargo fmt -- --write-mode=diff
-    - cargo build --features nightly
+    - ./scripts/build-all.sh
     - cargo test --features nightly
     - cargo bench --features nightly
 

Cargo.toml

@@ -18,19 +18,24 @@ license = "MIT"
 
 [dependencies]
 hyper = "^0.10.6"
-unicase = "^1.0"
-url = "^1.0"
-bitflags = "^0.8"
-rand = "^0.3"
-byteorder = "^1.0"
-sha1 = "^0.2"
-openssl = { version = "^0.9.10", optional = true }
-base64 = "^0.5"
-
-[dev-dependencies]
-serde_json = "^1.0"
+unicase = "1.0"
+url = "1.0"
+bitflags = "0.8"
+rand = "0.3"
+byteorder = "1.0"
+sha1 = "0.2"
+base64 = "0.5"
+futures = { version = "0.1", optional = true }
+tokio-core = { version = "0.1", optional = true }
+tokio-io = { version = "^0.1.2", optional = true }
+tokio-tls = { version = "0.1", optional = true }
+bytes = { version = "0.4", optional = true }
+native-tls = { version = "^0.1.2", optional = true }
 
 [features]
-default = ["ssl"]
-ssl = ["openssl"]
+default = ["sync", "sync-ssl", "async", "async-ssl"]
+sync = []
+sync-ssl = ["native-tls", "sync"]
+async = ["tokio-core", "tokio-io", "bytes", "futures"]
+async-ssl = ["native-tls", "tokio-tls", "async"]
 nightly = ["hyper/nightly"]

autobahn/client-results.json

@@ -1723,7 +1723,7 @@
          "reportfile": "rust_websocket_case_3_2.json"
       },
       "3.3": {
-         "behavior": "NON-STRICT",
+         "behavior": "OK",
          "behaviorClose": "OK",
          "duration": 2,
          "remoteCloseCode": null,
@@ -2773,28 +2773,28 @@
          "reportfile": "rust_websocket_case_6_3_2.json"
       },
       "6.4.1": {
-         "behavior": "NON-STRICT",
+         "behavior": "OK",
          "behaviorClose": "OK",
          "duration": 2002,
          "remoteCloseCode": null,
          "reportfile": "rust_websocket_case_6_4_1.json"
       },
       "6.4.2": {
-         "behavior": "NON-STRICT",
+         "behavior": "OK",
          "behaviorClose": "OK",
          "duration": 2002,
          "remoteCloseCode": null,
          "reportfile": "rust_websocket_case_6_4_2.json"
       },
       "6.4.3": {
-         "behavior": "NON-STRICT",
+         "behavior": "OK",
          "behaviorClose": "OK",
          "duration": 2003,
          "remoteCloseCode": null,
          "reportfile": "rust_websocket_case_6_4_3.json"
       },
       "6.4.4": {
-         "behavior": "NON-STRICT",
+         "behavior": "OK",
          "behaviorClose": "OK",
          "duration": 2002,
          "remoteCloseCode": null,
@@ -3634,4 +3634,4 @@
          "reportfile": "rust_websocket_case_9_8_6.json"
       }
    }
-}
+}

autobahn/server-results.json

@@ -2773,28 +2773,28 @@
          "reportfile": "rust_websocket_case_6_3_2.json"
       },
       "6.4.1": {
-         "behavior": "NON-STRICT",
+         "behavior": "OK",
          "behaviorClose": "OK",
          "duration": 2002,
          "remoteCloseCode": null,
          "reportfile": "rust_websocket_case_6_4_1.json"
       },
       "6.4.2": {
-         "behavior": "NON-STRICT",
+         "behavior": "OK",
          "behaviorClose": "OK",
          "duration": 2001,
          "remoteCloseCode": null,
          "reportfile": "rust_websocket_case_6_4_2.json"
       },
       "6.4.3": {
-         "behavior": "NON-STRICT",
+         "behavior": "OK",
          "behaviorClose": "OK",
          "duration": 2003,
          "remoteCloseCode": null,
          "reportfile": "rust_websocket_case_6_4_3.json"
       },
       "6.4.4": {
-         "behavior": "NON-STRICT",
+         "behavior": "OK",
          "behaviorClose": "OK",
          "duration": 2003,
          "remoteCloseCode": null,
@@ -3634,4 +3634,4 @@
          "reportfile": "rust_websocket_case_9_8_6.json"
       }
    }
-}
+}

examples/async-autobahn-client.rs

@@ -0,0 +1,109 @@
+extern crate websocket;
+extern crate tokio_core;
+extern crate futures;
+
+use websocket::{ClientBuilder, OwnedMessage};
+use websocket::result::WebSocketError;
+use tokio_core::reactor::Core;
+use futures::sink::Sink;
+use futures::stream::Stream;
+use futures::Future;
+use futures::future::{self, Loop};
+
+fn main() {
+	let addr = "ws://127.0.0.1:9001".to_string();
+	let agent = "rust-websocket";
+	let mut core = Core::new().unwrap();
+	let handle = core.handle();
+
+	println!("Using fuzzingserver {}", addr);
+	println!("Using agent {}", agent);
+
+	let case_count = get_case_count(addr.clone(), &mut core);
+	println!("We will be running {} test cases!", case_count);
+
+	println!("Running test suite...");
+	for case_id in 1..(case_count + 1) {
+		let url = addr.clone() + "/runCase?case=" + &case_id.to_string()[..] + "&agent=" + agent;
+
+		let test_case = ClientBuilder::new(&url)
+			.unwrap()
+			.async_connect_insecure(&handle)
+			.and_then(move |(duplex, _)| {
+				println!("Executing test case: {}/{}", case_id, case_count);
+				future::loop_fn(duplex, |stream| {
+					stream.into_future()
+					      .or_else(|(err, stream)| {
+						               println!("Could not receive message: {:?}", err);
+						               stream.send(OwnedMessage::Close(None)).map(|s| (None, s))
+						              })
+					      .and_then(|(msg, stream)| match msg {
+					                    Some(OwnedMessage::Text(txt)) => {
+						                    stream.send(OwnedMessage::Text(txt))
+					                              .map(|s| Loop::Continue(s))
+					                              .boxed()
+					                    }
+					                    Some(OwnedMessage::Binary(bin)) => {
+						                    stream.send(OwnedMessage::Binary(bin))
+					                              .map(|s| Loop::Continue(s))
+					                              .boxed()
+					                    }
+					                    Some(OwnedMessage::Ping(data)) => {
+						                    stream.send(OwnedMessage::Pong(data))
+					                              .map(|s| Loop::Continue(s))
+					                              .boxed()
+					                    }
+					                    Some(OwnedMessage::Close(_)) => {
+						                    stream.send(OwnedMessage::Close(None))
+					                              .map(|_| Loop::Break(()))
+					                              .boxed()
+					                    }
+					                    Some(OwnedMessage::Pong(_)) => {
+						                    future::ok(Loop::Continue(stream)).boxed()
+					                    }
+					                    None => future::ok(Loop::Break(())).boxed(),
+					                })
+				})
+			})
+			.map(move |_| {
+				     println!("Test case {} is finished!", case_id);
+				    })
+			.or_else(move |err| {
+				         println!("Test case {} ended with an error: {:?}", case_id, err);
+				         Ok(()) as Result<(), ()>
+				        });
+
+		core.run(test_case).ok();
+	}
+
+	update_reports(addr.clone(), agent, &mut core);
+	println!("Test suite finished!");
+}
+
+fn get_case_count(addr: String, core: &mut Core) -> usize {
+	let url = addr + "/getCaseCount";
+	let err = "Unsupported message in /getCaseCount";
+
+	let counter = ClientBuilder::new(&url)
+		.unwrap()
+		.async_connect_insecure(&core.handle())
+		.and_then(|(s, _)| s.into_future().map_err(|e| e.0))
+		.and_then(|(msg, _)| match msg {
+		              Some(OwnedMessage::Text(txt)) => Ok(txt.parse().unwrap()),
+		              _ => Err(WebSocketError::ProtocolError(err)),
+		          });
+	core.run(counter).unwrap()
+}
+
+fn update_reports(addr: String, agent: &str, core: &mut Core) {
+	println!("Updating reports...");
+	let url = addr + "/updateReports?agent=" + agent;
+
+	let updater = ClientBuilder::new(&url)
+		.unwrap()
+		.async_connect_insecure(&core.handle())
+		.and_then(|(sink, _)| sink.send(OwnedMessage::Close(None)));
+	core.run(updater).unwrap();
+
+	println!("Reports updated.");
+}

examples/async-autobahn-server.rs

@@ -0,0 +1,54 @@
+extern crate websocket;
+extern crate futures;
+extern crate tokio_core;
+
+use websocket::message::OwnedMessage;
+use websocket::server::InvalidConnection;
+use websocket::async::Server;
+
+use tokio_core::reactor::Core;
+use futures::{Future, Sink, Stream};
+
+fn main() {
+	let mut core = Core::new().unwrap();
+	let handle = core.handle();
+	// bind to the server
+	let server = Server::bind("127.0.0.1:9002", &handle).unwrap();
+
+	// time to build the server's future
+	// this will be a struct containing everything the server is going to do
+
+	// a stream of incoming connections
+	let f = server.incoming()
+        // we don't wanna save the stream if it drops
+        .map_err(|InvalidConnection { error, .. }| error)
+        .for_each(|(upgrade, addr)| {
+            // accept the request to be a ws connection
+            println!("Got a connection from: {}", addr);
+            let f = upgrade
+                .accept()
+                .and_then(|(s, _)| {
+                    // simple echo server impl
+                    let (sink, stream) = s.split();
+                    stream
+                    .take_while(|m| Ok(!m.is_close()))
+                    .filter_map(|m| {
+                        match m {
+                            OwnedMessage::Ping(p) => Some(OwnedMessage::Pong(p)),
+                            OwnedMessage::Pong(_) => None,
+                            _ => Some(m),
+                        }
+                    })
+                    .forward(sink)
+                    .and_then(|(_, sink)| {
+                        sink.send(OwnedMessage::Close(None))
+                    })
+                });
+
+	          handle.spawn(f.map_err(move |e| println!("{}: '{:?}'", addr, e))
+	                       .map(move |_| println!("{} closed.", addr)));
+            Ok(())
+        });
+
+	core.run(f).unwrap();
+}

examples/async-client.rs

@@ -0,0 +1,65 @@
+extern crate websocket;
+extern crate futures;
+extern crate tokio_core;
+
+use std::thread;
+use std::io::stdin;
+use tokio_core::reactor::Core;
+use futures::future::Future;
+use futures::sink::Sink;
+use futures::stream::Stream;
+use futures::sync::mpsc;
+use websocket::result::WebSocketError;
+use websocket::{ClientBuilder, OwnedMessage};
+
+const CONNECTION: &'static str = "ws://127.0.0.1:2794";
+
+fn main() {
+	println!("Connecting to {}", CONNECTION);
+	let mut core = Core::new().unwrap();
+
+	// standard in isn't supported in mio yet, so we use a thread
+	// see https://github.com/carllerche/mio/issues/321
+	let (usr_msg, stdin_ch) = mpsc::channel(0);
+	thread::spawn(move || {
+		let mut input = String::new();
+		let mut stdin_sink = usr_msg.wait();
+		loop {
+			input.clear();
+			stdin().read_line(&mut input).unwrap();
+			let trimmed = input.trim();
+
+			let (close, msg) = match trimmed {
+				"/close" => (true, OwnedMessage::Close(None)),
+				"/ping" => (false, OwnedMessage::Ping(b"PING".to_vec())),
+				_ => (false, OwnedMessage::Text(trimmed.to_string())),
+			};
+
+			stdin_sink.send(msg)
+			          .expect("Sending message across stdin channel.");
+
+			if close {
+				break;
+			}
+		}
+	});
+
+	let runner = ClientBuilder::new(CONNECTION)
+		.unwrap()
+		.add_protocol("rust-websocket")
+		.async_connect_insecure(&core.handle())
+		.and_then(|(duplex, _)| {
+			let (sink, stream) = duplex.split();
+			stream.filter_map(|message| {
+				                  println!("Received Message: {:?}", message);
+				                  match message {
+				                      OwnedMessage::Close(e) => Some(OwnedMessage::Close(e)),
+				                      OwnedMessage::Ping(d) => Some(OwnedMessage::Pong(d)),
+				                      _ => None,
+				                  }
+				                 })
+			      .select(stdin_ch.map_err(|_| WebSocketError::NoDataAvailable))
+			      .forward(sink)
+		});
+	core.run(runner).unwrap();
+}