Fixed bug in async SSL connections.

Author: illegalprime

examples/ssl-client.rs

@@ -0,0 +1,64 @@
+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 = "wss://echo.websocket.org";
+
+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()
+		.async_connect_secure(None, &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();
+}

src/client/builder.rs

@@ -490,7 +490,7 @@ impl<'u> ClientBuilder<'u> {
 		handle: &Handle,
 	) -> async::ClientNew<Box<stream::async::Stream + Send>> {
 		// connect to the tcp stream
-		let tcp_stream = match self.async_tcpstream(handle) {
+		let tcp_stream = match self.async_tcpstream(None, handle) {
 			Ok(t) => t,
 			Err(e) => return future::err(e).boxed(),
 		};
@@ -543,7 +543,7 @@ impl<'u> ClientBuilder<'u> {
 		handle: &Handle,
 	) -> async::ClientNew<async::TlsStream<async::TcpStream>> {
 		// connect to the tcp stream
-		let tcp_stream = match self.async_tcpstream(handle) {
+		let tcp_stream = match self.async_tcpstream(Some(true), handle) {
 			Ok(t) => t,
 			Err(e) => return future::err(e).boxed(),
 		};
@@ -565,22 +565,22 @@ impl<'u> ClientBuilder<'u> {
 		};
 
 		// put it all together
-		let future =
-			tcp_stream.map_err(|e| e.into())
-			          .and_then(move |s| {
-				                    connector.connect_async(&host, s).map_err(|e| e.into())
-				                   })
-			          .and_then(move |stream| {
-				                    builder.async_connect_on(stream)
-				                   });
+		let future = tcp_stream.map_err(|e| e.into())
+		                       .and_then(move |s| {
+			                                 connector.connect_async(&host, s)
+			                                          .map_err(|e| e.into())
+			                                })
+		                       .and_then(move |stream| {
+			                                 builder.async_connect_on(stream)
+			                                });
 		Box::new(future)
 	}
 
 	// TODO: add timeout option for connecting
 	// TODO: add conveniences like .response_to_pings, .send_close, etc.
 	#[cfg(feature="async")]
 	pub fn async_connect_insecure(self, handle: &Handle) -> async::ClientNew<async::TcpStream> {
-		let tcp_stream = match self.async_tcpstream(handle) {
+		let tcp_stream = match self.async_tcpstream(Some(false), handle) {
 			Ok(t) => t,
 			Err(e) => return future::err(e).boxed(),
 		};
@@ -645,20 +645,23 @@ impl<'u> ClientBuilder<'u> {
 	}
 
 	#[cfg(feature="async")]
-	fn async_tcpstream(&self, handle: &Handle) -> WebSocketResult<TcpStreamNew> {
+	fn async_tcpstream(
+		&self,
+		secure: Option<bool>,
+		handle: &Handle,
+	) -> WebSocketResult<TcpStreamNew> {
 		// get the address to connect to, return an error future if ther's a problem
-		let address =
-			match self.extract_host_port(Some(false)).and_then(|p| Ok(p.to_socket_addrs()?)) {
-				Ok(mut s) => {
-					match s.next() {
-						Some(a) => a,
-						None => {
-							return Err(WebSocketError::WebSocketUrlError(WSUrlErrorKind::NoHostName));
-						}
+		let address = match self.extract_host_port(secure).and_then(|p| Ok(p.to_socket_addrs()?)) {
+			Ok(mut s) => {
+				match s.next() {
+					Some(a) => a,
+					None => {
+						return Err(WebSocketError::WebSocketUrlError(WSUrlErrorKind::NoHostName));
 					}
 				}
-				Err(e) => return Err(e.into()),
-			};
+			}
+			Err(e) => return Err(e.into()),
+		};
 
 		// connect a tcp stream
 		Ok(async::TcpStream::connect(&address, handle))