1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
extern crate tokio_dns;
extern crate tokio_core;
use self::tokio_dns::tcp_connect;
use self::tokio_core::reactor::Remote;
use std::io::Result as IoResult;
use futures::{Future, BoxFuture};
use futures::future;
use super::{WebSocketStream, Request, client_async};
use tungstenite::Error;
use tungstenite::client::url_mode;
use stream::NoDelay;
use self::tokio_core::net::TcpStream;
impl NoDelay for TcpStream {
fn set_nodelay(&mut self, nodelay: bool) -> IoResult<()> {
TcpStream::set_nodelay(self, nodelay)
}
}
#[cfg(feature="tls")]
mod encryption {
extern crate native_tls;
extern crate tokio_tls;
use super::tokio_core::net::TcpStream;
use self::native_tls::TlsConnector;
use self::tokio_tls::{TlsConnectorExt, TlsStream};
use std::io::{Read, Write, Result as IoResult};
use futures::{Future, BoxFuture};
use futures::future;
use tungstenite::Error;
use tungstenite::stream::Mode;
use stream::NoDelay;
pub use stream::Stream as StreamSwitcher;
pub type AutoStream = StreamSwitcher<TcpStream, TlsStream<TcpStream>>;
impl<T: Read + Write + NoDelay> NoDelay for TlsStream<T> {
fn set_nodelay(&mut self, nodelay: bool) -> IoResult<()> {
self.get_mut().get_mut().set_nodelay(nodelay)
}
}
pub fn wrap_stream(socket: TcpStream, domain: String, mode: Mode) -> BoxFuture<AutoStream, Error> {
match mode {
Mode::Plain => future::ok(StreamSwitcher::Plain(socket)).boxed(),
Mode::Tls => {
future::result(TlsConnector::builder())
.and_then(move |builder| future::result(builder.build()))
.and_then(move |connector| connector.connect_async(&domain, socket))
.map(|s| StreamSwitcher::Tls(s))
.map_err(|e| Error::Tls(e))
.boxed()
}
}
}
}
#[cfg(not(feature="tls"))]
mod encryption {
use super::tokio_core::net::TcpStream;
use futures::{Future, BoxFuture};
use futures::future;
use tungstenite::Error;
use tungstenite::stream::Mode;
pub type AutoStream = TcpStream;
pub fn wrap_stream(socket: TcpStream, _domain: String, mode: Mode) -> BoxFuture<AutoStream, Error> {
match mode {
Mode::Plain => future::ok(socket).boxed(),
Mode::Tls => future::err(Error::Url("TLS support not compiled in.".into())).boxed(),
}
}
}
use self::encryption::{AutoStream, wrap_stream};
pub fn connect_async<R>(request: R, handle: Remote) -> BoxFuture<WebSocketStream<AutoStream>, Error>
where R: Into<Request<'static>>
{
let request: Request = request.into();
let mode = match url_mode(&request.url) {
Ok(m) => m,
Err(e) => return future::err(e.into()).boxed(),
};
let domain = match request.url.host_str() {
Some(d) => d.to_string(),
None => return future::err(Error::Url("No host name in the URL".into())).boxed(),
};
let port = request.url.port_or_known_default().expect("Bug: port unknown");
tcp_connect((domain.as_str(), port), handle).map_err(|e| e.into())
.and_then(move |socket| wrap_stream(socket, domain, mode))
.and_then(|mut stream| {
NoDelay::set_nodelay(&mut stream, true)
.map(move |()| stream)
.map_err(|e| e.into())
})
.and_then(move |stream| client_async(request, stream))
.boxed()
}