use std::cell::RefCell;
use std::sync::mpsc;
use std::{net, thread, time};
use actix_codec::{AsyncRead, AsyncWrite, Framed};
use actix_rt::{Runtime, System};
use actix_server::{Server, StreamServiceFactory};
use awc::{error::PayloadError, ws, Client, ClientRequest, ClientResponse, Connector};
use bytes::Bytes;
use futures::future::lazy;
use futures::{Future, Stream};
use http::Method;
use net2::TcpBuilder;
thread_local! {
static RT: RefCell<Runtime> = {
RefCell::new(Runtime::new().unwrap())
};
}
pub fn block_on<F>(f: F) -> Result<F::Item, F::Error>
where
F: Future,
{
RT.with(move |rt| rt.borrow_mut().block_on(f))
}
pub fn run_on<F, R>(f: F) -> R
where
F: Fn() -> R,
{
RT.with(move |rt| rt.borrow_mut().block_on(lazy(|| Ok::<_, ()>(f()))))
.unwrap()
}
pub struct TestServer;
pub struct TestServerRuntime {
addr: net::SocketAddr,
rt: Runtime,
client: Client,
}
impl TestServer {
pub fn new<F: StreamServiceFactory>(factory: F) -> TestServerRuntime {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
let sys = System::new("actix-test-server");
let tcp = net::TcpListener::bind("127.0.0.1:0").unwrap();
let local_addr = tcp.local_addr().unwrap();
Server::build()
.listen("test", tcp, factory)?
.workers(1)
.disable_signals()
.start();
tx.send((System::current(), local_addr)).unwrap();
sys.run()
});
let (system, addr) = rx.recv().unwrap();
let mut rt = Runtime::new().unwrap();
let client = rt
.block_on(lazy(move || {
let connector = {
#[cfg(feature = "ssl")]
{
use openssl::ssl::{SslConnector, SslMethod, SslVerifyMode};
let mut builder =
SslConnector::builder(SslMethod::tls()).unwrap();
builder.set_verify(SslVerifyMode::NONE);
let _ = builder.set_alpn_protos(b"\x02h2\x08http/1.1").map_err(
|e| log::error!("Can not set alpn protocol: {:?}", e),
);
Connector::new()
.conn_lifetime(time::Duration::from_secs(0))
.timeout(time::Duration::from_millis(500))
.ssl(builder.build())
.finish()
}
#[cfg(not(feature = "ssl"))]
{
Connector::new()
.conn_lifetime(time::Duration::from_secs(0))
.timeout(time::Duration::from_millis(500))
.finish()
}
};
Ok::<Client, ()>(Client::build().connector(connector).finish())
}))
.unwrap();
System::set_current(system);
TestServerRuntime { addr, rt, client }
}
pub fn unused_addr() -> net::SocketAddr {
let addr: net::SocketAddr = "127.0.0.1:0".parse().unwrap();
let socket = TcpBuilder::new_v4().unwrap();
socket.bind(&addr).unwrap();
socket.reuse_address(true).unwrap();
let tcp = socket.to_tcp_listener().unwrap();
tcp.local_addr().unwrap()
}
}
impl TestServerRuntime {
pub fn block_on<F, I, E>(&mut self, fut: F) -> Result<I, E>
where
F: Future<Item = I, Error = E>,
{
self.rt.block_on(fut)
}
pub fn block_on_fn<F, R>(&mut self, f: F) -> Result<R::Item, R::Error>
where
F: FnOnce() -> R,
R: Future,
{
self.rt.block_on(lazy(|| f()))
}
pub fn execute<F, R>(&mut self, fut: F) -> R
where
F: FnOnce() -> R,
{
self.rt.block_on(lazy(|| Ok::<_, ()>(fut()))).unwrap()
}
pub fn addr(&self) -> net::SocketAddr {
self.addr
}
pub fn url(&self, uri: &str) -> String {
if uri.starts_with('/') {
format!("http://127.0.0.1:{}{}", self.addr.port(), uri)
} else {
format!("http://127.0.0.1:{}/{}", self.addr.port(), uri)
}
}
pub fn surl(&self, uri: &str) -> String {
if uri.starts_with('/') {
format!("https://127.0.0.1:{}{}", self.addr.port(), uri)
} else {
format!("https://127.0.0.1:{}/{}", self.addr.port(), uri)
}
}
pub fn get<S: AsRef<str>>(&self, path: S) -> ClientRequest {
self.client.get(self.url(path.as_ref()).as_str())
}
pub fn sget<S: AsRef<str>>(&self, path: S) -> ClientRequest {
self.client.get(self.surl(path.as_ref()).as_str())
}
pub fn post<S: AsRef<str>>(&self, path: S) -> ClientRequest {
self.client.post(self.url(path.as_ref()).as_str())
}
pub fn spost<S: AsRef<str>>(&self, path: S) -> ClientRequest {
self.client.post(self.surl(path.as_ref()).as_str())
}
pub fn head<S: AsRef<str>>(&self, path: S) -> ClientRequest {
self.client.head(self.url(path.as_ref()).as_str())
}
pub fn shead<S: AsRef<str>>(&self, path: S) -> ClientRequest {
self.client.head(self.surl(path.as_ref()).as_str())
}
pub fn request<S: AsRef<str>>(&self, method: Method, path: S) -> ClientRequest {
self.client.request(method, path.as_ref())
}
pub fn load_body<S>(
&mut self,
mut response: ClientResponse<S>,
) -> Result<Bytes, PayloadError>
where
S: Stream<Item = Bytes, Error = PayloadError> + 'static,
{
self.block_on(response.body().limit(10_485_760))
}
pub fn ws_at(
&mut self,
path: &str,
) -> Result<Framed<impl AsyncRead + AsyncWrite, ws::Codec>, awc::error::WsClientError>
{
let url = self.url(path);
let connect = self.client.ws(url).connect();
self.rt
.block_on(lazy(move || connect.map(|(_, framed)| framed)))
}
pub fn ws(
&mut self,
) -> Result<Framed<impl AsyncRead + AsyncWrite, ws::Codec>, awc::error::WsClientError>
{
self.ws_at("/")
}
fn stop(&mut self) {
System::current().stop();
}
}
impl Drop for TestServerRuntime {
fn drop(&mut self) {
self.stop()
}
}