berkus
/
actix-net
mirror of https://github.com/fafhrd91/actix-net
1
0
Fork 0
Code Issues Releases Wiki Activity

merged

This commit is contained in:
Jerome Gravel-Niquet 2019-09-27 11:09:49 -04:00
parent 25b2529396 aa9bbe2114
commit 551aab974a
24 changed files with 417 additions and 925 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
CHANGES.md
View File

@ -1,68 +0,0 @@
# Changes
## [0.3.0] - xxx
* Split `Service` trait to separate crate
* Use new `Service<Request>` trait
## [0.2.4] - 2018-11-21
### Added
* Allow to skip name resolution stage in Connector
## [0.2.3] - 2018-11-17
### Added
* Framed::is_write_buf_empty() checks if write buffer is flushed
## [0.2.2] - 2018-11-14
### Added
* Add low/high caps to Framed
### Changed
* Refactor Connector and Resolver services
### Fixed
* Fix wrong service to socket binding
## [0.2.0] - 2018-11-08
### Added
* Timeout service
* Added ServiceConfig and ServiceRuntime for server service configuration
### Changed
* Connector has been refactored
* timer and LowResTimer renamed to time and LowResTime
* Refactored `Server::configure()` method
## [0.1.1] - 2018-10-10
### Changed
- Set actix min version - 0.7.5
- Set trust-dns min version
## [0.1.0] - 2018-10-08
* Initial impl

20
Cargo.toml
View File

@ -17,14 +17,13 @@ edition = "2018"
members = [
"actix-codec",
"actix-connect",
"actix-ioframe",
"actix-rt",
"actix-service",
"actix-server",
"actix-server-config",
"actix-test-server",
"actix-testing",
"actix-threadpool",
"actix-tower",
"actix-ioframe",
"actix-utils",
"router",
]
@ -38,4 +37,17 @@ env_logger = "0.6"
futures = "0.1.25"
openssl = "0.10"
tokio-tcp = "0.1"
tokio-openssl = "0.3"
tokio-openssl = "0.3"
[patch.crates-io]
actix-codec = { path = "actix-codec" }
actix-connect = { path = "actix-connect" }
actix-ioframe = { path = "actix-ioframe" }
actix-rt = { path = "actix-rt" }
actix-server = { path = "actix-server" }
actix-server-config = { path = "actix-server-config" }
actix-service = { path = "actix-service" }
actix-testing = { path = "actix-testing" }
actix-threadpool = { path = "actix-threadpool" }
actix-utils = { path = "actix-utils" }
actix-router = { path = "router" }

2
README.md
View File

@ -1,4 +1,4 @@
# Actix net [![Build Status](https://travis-ci.org/actix/actix-net.svg?branch=master)](https://travis-ci.org/actix/actix-net) [![codecov](https://codecov.io/gh/actix/actix-net/branch/master/graph/badge.svg)](https://codecov.io/gh/actix/actix-net) [![crates.io](https://meritbadge.herokuapp.com/actix-net)](https://crates.io/crates/actix-net) [![Join the chat at https://gitter.im/actix/actix](https://badges.gitter.im/actix/actix.svg)](https://gitter.im/actix/actix?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
# Actix net [![Build Status](https://travis-ci.org/actix/actix-net.svg?branch=master)](https://travis-ci.org/actix/actix-net) [![codecov](https://codecov.io/gh/actix/actix-net/branch/master/graph/badge.svg)](https://codecov.io/gh/actix/actix-net) [![Join the chat at https://gitter.im/actix/actix](https://badges.gitter.im/actix/actix.svg)](https://gitter.im/actix/actix?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
Actix net - framework for composable network services

2
actix-connect/Cargo.toml
View File

@ -57,5 +57,5 @@ webpki = { version = "0.21", optional = true }
[dev-dependencies]
bytes = "0.4"
actix-test-server = { version="0.2.2", features=["ssl"] }
actix-testing = { version="0.1.0" }
actix-server-config = "0.1.0"

82
actix-connect/tests/test_connect.rs
View File

@ -1,7 +1,7 @@
use actix_codec::{BytesCodec, Framed};
use actix_server_config::Io;
use actix_service::{service_fn, NewService, Service};
use actix_test_server::TestServer;
use actix_testing::{self as test, TestServer};
use bytes::Bytes;
use futures::{future::lazy, Future, Sink};
use http::{HttpTryFrom, Uri};
@ -12,7 +12,7 @@ use actix_connect::{default_connector, Connect};
#[cfg(feature = "ssl")]
#[test]
fn test_string() {
let mut srv = TestServer::with(|| {
let srv = TestServer::with(|| {
service_fn(|io: Io<tokio_tcp::TcpStream>| {
Framed::new(io.into_parts().0, BytesCodec)
.send(Bytes::from_static(b"test"))
@ -22,14 +22,14 @@ fn test_string() {
let mut conn = default_connector();
let addr = format!("localhost:{}", srv.port());
let con = srv.run_on(move || conn.call(addr.into())).unwrap();
let con = test::call_service(&mut conn, addr.into());
assert_eq!(con.peer_addr().unwrap(), srv.addr());
}
#[cfg(feature = "rust-tls")]
#[test]
fn test_rustls_string() {
let mut srv = TestServer::with(|| {
let srv = TestServer::with(|| {
service_fn(|io: Io<tokio_tcp::TcpStream>| {
Framed::new(io.into_parts().0, BytesCodec)
.send(Bytes::from_static(b"test"))
@ -39,12 +39,12 @@ fn test_rustls_string() {
let mut conn = default_connector();
let addr = format!("localhost:{}", srv.port());
let con = srv.run_on(move || conn.call(addr.into())).unwrap();
let con = test::call_service(&mut conn, addr.into());
assert_eq!(con.peer_addr().unwrap(), srv.addr());
}
#[test]
fn test_static_str() {
let mut srv = TestServer::with(|| {
let srv = TestServer::with(|| {
service_fn(|io: Io<tokio_tcp::TcpStream>| {
Framed::new(io.into_parts().0, BytesCodec)
.send(Bytes::from_static(b"test"))
@ -52,33 +52,29 @@ fn test_static_str() {
})
});
let resolver = srv
.block_on(lazy(
|| Ok::<_, ()>(actix_connect::start_default_resolver()),
))
.unwrap();
let mut conn = srv
.block_on(lazy(|| {
Ok::<_, ()>(actix_connect::new_connector(resolver.clone()))
}))
.unwrap();
let resolver = test::block_on(lazy(
|| Ok::<_, ()>(actix_connect::start_default_resolver()),
))
.unwrap();
let con = srv
.block_on(conn.call(Connect::with("10", srv.addr())))
.unwrap();
let mut conn = test::block_on(lazy(|| {
Ok::<_, ()>(actix_connect::new_connector(resolver.clone()))
}))
.unwrap();
let con = test::block_on(conn.call(Connect::with("10", srv.addr()))).unwrap();
assert_eq!(con.peer_addr().unwrap(), srv.addr());
let connect = Connect::new(srv.host().to_owned());
let mut conn = srv
.block_on(lazy(|| Ok::<_, ()>(actix_connect::new_connector(resolver))))
.unwrap();
let con = srv.block_on(conn.call(connect));
let mut conn =
test::block_on(lazy(|| Ok::<_, ()>(actix_connect::new_connector(resolver)))).unwrap();
let con = test::block_on(conn.call(connect));
assert!(con.is_err());
}
#[test]
fn test_new_service() {
let mut srv = TestServer::with(|| {
let srv = TestServer::with(|| {
service_fn(|io: Io<tokio_tcp::TcpStream>| {
Framed::new(io.into_parts().0, BytesCodec)
.send(Bytes::from_static(b"test"))
@ -86,31 +82,27 @@ fn test_new_service() {
})
});
let resolver = srv
.block_on(lazy(|| {
Ok::<_, ()>(actix_connect::start_resolver(
ResolverConfig::default(),
ResolverOpts::default(),
))
}))
.unwrap();
let factory = srv
.block_on(lazy(|| {
Ok::<_, ()>(actix_connect::new_connector_factory(resolver))
}))
.unwrap();
let resolver = test::block_on(lazy(|| {
Ok::<_, ()>(actix_connect::start_resolver(
ResolverConfig::default(),
ResolverOpts::default(),
))
}))
.unwrap();
let factory = test::block_on(lazy(|| {
Ok::<_, ()>(actix_connect::new_connector_factory(resolver))
}))
.unwrap();
let mut conn = srv.block_on(factory.new_service(&())).unwrap();
let con = srv
.block_on(conn.call(Connect::with("10", srv.addr())))
.unwrap();
let mut conn = test::block_on(factory.new_service(&())).unwrap();
let con = test::block_on(conn.call(Connect::with("10", srv.addr()))).unwrap();
assert_eq!(con.peer_addr().unwrap(), srv.addr());
}
#[cfg(feature = "ssl")]
#[test]
fn test_uri() {
let mut srv = TestServer::with(|| {
let srv = TestServer::with(|| {
service_fn(|io: Io<tokio_tcp::TcpStream>| {
Framed::new(io.into_parts().0, BytesCodec)
.send(Bytes::from_static(b"test"))
@ -120,14 +112,14 @@ fn test_uri() {
let mut conn = default_connector();
let addr = Uri::try_from(format!("https://localhost:{}", srv.port())).unwrap();
let con = srv.run_on(move || conn.call(addr.into())).unwrap();
let con = test::call_service(&mut conn, addr.into());
assert_eq!(con.peer_addr().unwrap(), srv.addr());
}
#[cfg(feature = "rust-tls")]
#[test]
fn test_rustls_uri() {
let mut srv = TestServer::with(|| {
let srv = TestServer::with(|| {
service_fn(|io: Io<tokio_tcp::TcpStream>| {
Framed::new(io.into_parts().0, BytesCodec)
.send(Bytes::from_static(b"test"))
@ -137,6 +129,6 @@ fn test_rustls_uri() {
let mut conn = default_connector();
let addr = Uri::try_from(format!("https://localhost:{}", srv.port())).unwrap();
let con = srv.run_on(move || conn.call(addr.into())).unwrap();
let con = test::call_service(&mut conn, addr.into());
assert_eq!(con.peer_addr().unwrap(), srv.addr());
}

2
actix-ioframe/CHANGES.md
View File

@ -1,5 +1,5 @@
# Changes
## [0.1.0] - 2019-07-17
## [0.1.0] - 2019-09-25
* Initial release

4
actix-ioframe/Cargo.toml
View File

@ -6,7 +6,7 @@ description = "Actix framed service"
keywords = ["network", "framework", "async", "futures"]
homepage = "https://actix.rs"
repository = "https://github.com/actix/actix-net.git"
documentation = "https://docs.rs/actix-ioframed/"
documentation = "https://docs.rs/actix-ioframe/"
categories = ["network-programming", "asynchronous"]
license = "MIT/Apache-2.0"
exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
@ -29,7 +29,7 @@ log = "0.4"
[dev-dependencies]
actix-rt = "0.2.2"
actix-connect = "0.2.0"
actix-test-server = "0.2.2"
actix-testing = "0.1.0"
actix-server-config = "0.1.1"
tokio-tcp = "0.1"
tokio-timer = "0.2"

19
actix-ioframe/tests/test_server.rs
View File

@ -5,7 +5,7 @@ use std::time::Duration;
use actix_codec::BytesCodec;
use actix_server_config::Io;
use actix_service::{new_apply_fn, Service};
use actix_test_server::TestServer;
use actix_testing::{self as test, TestServer};
use futures::Future;
use tokio_tcp::TcpStream;
use tokio_timer::sleep;
@ -19,7 +19,7 @@ fn test_disconnect() -> std::io::Result<()> {
let disconnect = Arc::new(AtomicBool::new(false));
let disconnect1 = disconnect.clone();
let mut srv = TestServer::with(move || {
let srv = TestServer::with(move || {
let disconnect1 = disconnect1.clone();
new_apply_fn(
@ -41,15 +41,14 @@ fn test_disconnect() -> std::io::Result<()> {
})
.finish(|_t| Ok(None));
let conn = srv
.block_on(
actix_connect::default_connector()
.call(actix_connect::Connect::with(String::new(), srv.addr())),
)
.unwrap();
let conn = test::block_on(
actix_connect::default_connector()
.call(actix_connect::Connect::with(String::new(), srv.addr())),
)
.unwrap();
srv.block_on(client.call(conn.into_parts().0)).unwrap();
let _ = srv.block_on(
test::block_on(client.call(conn.into_parts().0)).unwrap();
let _ = test::block_on(
sleep(Duration::from_millis(100))
.map(|_| ())
.map_err(|_| ()),

7
actix-server/CHANGES.md
View File

@ -1,5 +1,12 @@
# Changes
## [0.6.1] - 2019-09-25
### Added
* Add UDS listening support to `ServerBuilder`
## [0.6.0] - 2019-07-18
### Added

2
actix-server/Cargo.toml
View File

@ -1,6 +1,6 @@
[package]
name = "actix-server"
version = "0.6.0"
version = "0.6.1"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix server - General purpose tcp server"
keywords = ["network", "framework", "async", "futures"]

29
actix-server/src/builder.rs
View File

@ -185,20 +185,41 @@ impl ServerBuilder {
#[cfg(all(unix, feature = "uds"))]
/// Add new unix domain service to the server.
pub fn bind_uds<F, U, N>(mut self, name: N, addr: U, factory: F) -> io::Result<Self>
pub fn bind_uds<F, U, N>(self, name: N, addr: U, factory: F) -> io::Result<Self>
where
F: ServiceFactory<tokio_uds::UnixStream>,
N: AsRef<str>,
U: AsRef<std::path::Path>,
{
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::os::unix::net::UnixListener;
// TODO: need to do something with existing paths
let _ = std::fs::remove_file(addr.as_ref());
// The path must not exist when we try to bind.
// Try to remove it to avoid bind error.
if let Err(e) = std::fs::remove_file(addr.as_ref()) {
// NotFound is expected and not an issue. Anything else is.
if e.kind() != std::io::ErrorKind::NotFound {
return Err(e);
}
}
let lst = UnixListener::bind(addr)?;
self.listen_uds(name, lst, factory)
}
#[cfg(all(unix, feature = "uds"))]
/// Add new unix domain service to the server.
/// Useful when running as a systemd service and
/// a socket FD can be acquired using the systemd crate.
pub fn listen_uds<F, N: AsRef<str>>(
mut self,
name: N,
lst: std::os::unix::net::UnixListener,
factory: F,
) -> io::Result<Self>
where
F: ServiceFactory<tokio_uds::UnixStream>,
{
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
let token = self.token.next();
let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8080);
self.services.push(StreamNewService::create(

31
actix-test-server/Cargo.toml
View File

@ -1,6 +1,6 @@
[package]
name = "actix-test-server"
version = "0.2.2"
version = "0.2.3"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix test server"
keywords = ["network", "framework", "async", "futures"]
@ -11,31 +11,16 @@ categories = ["network-programming", "asynchronous"]
license = "MIT/Apache-2.0"
exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
edition = "2018"
workspace = ".."
[package.metadata.docs.rs]
features = ["ssl", "tls", "rust-tls"]
[lib]
name = "actix_test_server"
path = "src/lib.rs"
[features]
default = []
# tls
tls = ["native-tls", "actix-server/tls"]
# openssl
ssl = ["openssl", "actix-server/ssl"]
# rustls
rust-tls = ["rustls", "tokio-rustls", "webpki", "webpki-roots"]
[dependencies]
actix-rt = "0.2.1"
actix-server = "0.5.0"
actix-server-config = "0.1.0"
actix-testing = "0.1.0"
log = "0.4"
net2 = "0.2"
@ -43,17 +28,5 @@ futures = "0.1"
tokio-tcp = "0.1"
tokio-reactor = "0.1"
# native-tls
native-tls = { version="0.2", optional = true }
# openssl
openssl = { version="0.10", optional = true }
#rustls
rustls = { version = "^0.15", optional = true }
tokio-rustls = { version = "^0.9", optional = true }
webpki = { version = "0.19", optional = true }
webpki-roots = { version = "0.16", optional = true }
[dev-dependencies]
actix-service = "0.4.0"

3
actix-test-server/README.md Normal file
View File

@ -0,0 +1,3 @@
# Actix test server (Deprecated)
Use [actix-testing](https://docs.rs/actix-testing/) instead

149
actix-test-server/src/lib.rs
View File

@ -1,149 +1,2 @@
//! Various helpers for Actix applications to use during testing.
use std::sync::mpsc;
use std::{net, thread};
use actix_rt::{Runtime, System};
use actix_server::{Server, StreamServiceFactory};
pub use actix_server_config::{Io, ServerConfig};
use futures::future::{lazy, Future, IntoFuture};
use net2::TcpBuilder;
use tokio_reactor::Handle;
use tokio_tcp::TcpStream;
/// The `TestServer` type.
///
/// `TestServer` is very simple test server that simplify process of writing
/// integration tests for actix-net applications.
///
/// # Examples
///
/// ```rust
/// use actix_service::{service_fn, IntoNewService};
/// use actix_test_server::TestServer;
///
/// fn main() {
/// let srv = TestServer::with(|| service_fn(
/// |sock| {
/// println!("New connection: {:?}", sock);
/// Ok::<_, ()>(())
/// }
/// ));
///
/// println!("SOCKET: {:?}", srv.connect());
/// }
/// ```
pub struct TestServer;
/// Test server runstime
pub struct TestServerRuntime {
addr: net::SocketAddr,
host: String,
port: u16,
rt: Runtime,
}
impl TestServer {
/// Start new test server with application factory
pub fn with<F: StreamServiceFactory>(factory: F) -> TestServerRuntime {
let (tx, rx) = mpsc::channel();
// run server in separate thread
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();
System::set_current(system);
let rt = Runtime::new().unwrap();
let host = format!("{}", addr.ip());
let port = addr.port();
TestServerRuntime {
addr,
rt,
host,
port,
}
}
/// Get firat available unused local address
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 {
/// Execute future on current runtime
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)
}
/// Runs the provided function, with runtime enabled.
pub fn run_on<F, R>(&mut self, f: F) -> Result<R::Item, R::Error>
where
F: FnOnce() -> R,
R: IntoFuture,
{
self.rt.block_on(lazy(|| f().into_future()))
}
/// Spawn future to the current runtime
pub fn spawn<F>(&mut self, fut: F)
where
F: Future<Item = (), Error = ()> + 'static,
{
self.rt.spawn(fut);
}
/// Test server host
pub fn host(&self) -> &str {
&self.host
}
/// Test server port
pub fn port(&self) -> u16 {
self.port
}
/// Get test server address
pub fn addr(&self) -> net::SocketAddr {
self.addr
}
/// Stop http server
fn stop(&mut self) {
System::current().stop();
}
/// Connect to server, return tokio TcpStream
pub fn connect(&self) -> std::io::Result<TcpStream> {
TcpStream::from_std(net::TcpStream::connect(self.addr)?, &Handle::default())
}
}
impl Drop for TestServerRuntime {
fn drop(&mut self) {
self.stop()
}
}
pub use actix_testing::*;

5
actix-testing/CHANGES.md Normal file
View File

@ -0,0 +1,5 @@
# Changes
## [0.1.0] - 2019-09-25
* Initial impl

29
actix-testing/Cargo.toml Normal file
View File

@ -0,0 +1,29 @@
[package]
name = "actix-testing"
version = "0.1.0"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix testing utils"
keywords = ["network", "framework", "async", "futures"]
homepage = "https://actix.rs"
repository = "https://github.com/actix/actix-net.git"
documentation = "https://docs.rs/actix-testing/"
categories = ["network-programming", "asynchronous"]
license = "MIT/Apache-2.0"
edition = "2018"
workspace = ".."
[lib]
name = "actix_testing"
path = "src/lib.rs"
[dependencies]
actix-rt = "0.2.1"
actix-server = "0.6.0"
actix-server-config = "0.1.0"
actix-service = "0.4.0"
log = "0.4"
net2 = "0.2"
futures = "0.1"
tokio-tcp = "0.1"
tokio-reactor = "0.1"

0
actix-tower/LICENSE-APACHE → actix-testing/LICENSE-APACHE
View File

0
actix-tower/LICENSE-MIT → actix-testing/LICENSE-MIT
View File

9
actix-testing/README.md Normal file
View File

@ -0,0 +1,9 @@
# Actix test utilities [![crates.io](https://meritbadge.herokuapp.com/actix-testing)](https://crates.io/crates/actix-testint) [![Join the chat at https://gitter.im/actix/actix](https://badges.gitter.im/actix/actix.svg)](https://gitter.im/actix/actix?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
## Documentation & community resources
* [User Guide](https://actix.rs/docs/)
* [API Documentation](https://docs.rs/actix-testing/)
* [Chat on gitter](https://gitter.im/actix/actix)
* Cargo package: [actix-http-test](https://crates.io/crates/actix-testing)
* Minimum supported Rust version: 1.37 or later

152
actix-testing/src/lib.rs Normal file
View File

@ -0,0 +1,152 @@
//! Various helpers for Actix applications to use during testing.
use std::sync::mpsc;
use std::{net, thread};
use actix_rt::System;
use actix_server::{Server, ServerBuilder, StreamServiceFactory};
pub use actix_server_config::{Io, ServerConfig};
use net2::TcpBuilder;
use tokio_reactor::Handle;
use tokio_tcp::TcpStream;
mod rt;
pub use self::rt::*;
/// The `TestServer` type.
///
/// `TestServer` is very simple test server that simplify process of writing
/// integration tests for actix-net applications.
///
/// # Examples
///
/// ```rust
/// use actix_service::{service_fn, IntoNewService};
/// use actix_testing::TestServer;
///
/// fn main() {
/// let srv = TestServer::with(|| service_fn(
/// |sock| {
/// println!("New connection: {:?}", sock);
/// Ok::<_, ()>(())
/// }
/// ));
///
/// println!("SOCKET: {:?}", srv.connect());
/// }
/// ```
pub struct TestServer;
/// Test server runstime
pub struct TestServerRuntime {
addr: net::SocketAddr,
host: String,
port: u16,
system: System,
}
impl TestServer {
/// Start new server with server builder
pub fn new<F>(mut factory: F) -> TestServerRuntime
where
F: FnMut(ServerBuilder) -> ServerBuilder + Send + 'static,
{
let (tx, rx) = mpsc::channel();
// run server in separate thread
thread::spawn(move || {
let sys = System::new("actix-test-server");
factory(Server::build())
.workers(1)
.disable_signals()
.start();
tx.send(System::current()).unwrap();
sys.run()
});
let system = rx.recv().unwrap();
TestServerRuntime {
system,
addr: "127.0.0.1:0".parse().unwrap(),
host: "127.0.0.1".to_string(),
port: 0,
}
}
/// Start new test server with application factory
pub fn with<F: StreamServiceFactory<TcpStream>>(factory: F) -> TestServerRuntime {
let (tx, rx) = mpsc::channel();
// run server in separate thread
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 host = format!("{}", addr.ip());
let port = addr.port();
TestServerRuntime {
system,
addr,
host,
port,
}
}
/// Get firat available unused local address
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 {
/// Test server host
pub fn host(&self) -> &str {
&self.host
}
/// Test server port
pub fn port(&self) -> u16 {
self.port
}
/// Get test server address
pub fn addr(&self) -> net::SocketAddr {
self.addr
}
/// Stop http server
fn stop(&mut self) {
self.system.stop();
}
/// Connect to server, return tokio TcpStream
pub fn connect(&self) -> std::io::Result<TcpStream> {
TcpStream::from_std(net::TcpStream::connect(self.addr)?, &Handle::default())
}
}
impl Drop for TestServerRuntime {
fn drop(&mut self) {
self.stop()
}
}

116
actix-testing/src/rt.rs Normal file
View File

@ -0,0 +1,116 @@
//! Various helpers for Actix applications to use during testing.
use std::cell::RefCell;
use actix_rt::{System, SystemRunner};
use actix_service::Service;
use futures::future::{lazy, Future, IntoFuture};
thread_local! {
static RT: RefCell<Inner> = {
RefCell::new(Inner(Some(System::builder().build())))
};
}
struct Inner(Option<SystemRunner>);
impl Inner {
fn get_mut(&mut self) -> &mut SystemRunner {
self.0.as_mut().unwrap()
}
}
impl Drop for Inner {
fn drop(&mut self) {
std::mem::forget(self.0.take().unwrap())
}
}
/// Runs the provided future, blocking the current thread until the future
/// completes.
///
/// This function can be used to synchronously block the current thread
/// until the provided `future` has resolved either successfully or with an
/// error. The result of the future is then returned from this function
/// call.
///
/// Note that this function is intended to be used only for testing purpose.
/// This function panics on nested call.
pub fn block_on<F>(f: F) -> Result<F::Item, F::Error>
where
F: IntoFuture,
{
RT.with(move |rt| rt.borrow_mut().get_mut().block_on(f.into_future()))
}
/// Runs the provided function, blocking the current thread until the result
/// future completes.
///
/// This function can be used to synchronously block the current thread
/// until the provided `future` has resolved either successfully or with an
/// error. The result of the future is then returned from this function
/// call.
///
/// Note that this function is intended to be used only for testing purpose.
/// This function panics on nested call.
pub fn block_fn<F, R>(f: F) -> Result<R::Item, R::Error>
where
F: FnOnce() -> R,
R: IntoFuture,
{
RT.with(move |rt| rt.borrow_mut().get_mut().block_on(lazy(f)))
}
/// Spawn future to the current test runtime.
pub fn spawn<F>(fut: F)
where
F: Future<Item = (), Error = ()> + 'static,
{
run_on(move || {
actix_rt::spawn(fut);
});
}
/// Runs the provided function, with runtime enabled.
///
/// Note that this function is intended to be used only for testing purpose.
/// This function panics on nested call.
pub fn run_on<F, R>(f: F) -> R
where
F: FnOnce() -> R,
{
RT.with(move |rt| {
rt.borrow_mut()
.get_mut()
.block_on(lazy(|| Ok::<_, ()>(f())))
})
.unwrap()
}
/// Calls service and waits for response future completion.
///
/// ```rust,ignore
/// use actix_web::{test, App, HttpResponse, http::StatusCode};
/// use actix_service::Service;
///
/// #[test]
/// fn test_response() {
/// let mut app = test::init_service(
/// App::new()
/// .service(web::resource("/test").to(|| HttpResponse::Ok()))
/// );
///
/// // Create request object
/// let req = test::TestRequest::with_uri("/test").to_request();
///
/// // Call application
/// let resp = test::call_service(&mut app, req);
/// assert_eq!(resp.status(), StatusCode::OK);
/// }
/// ```
pub fn call_service<S, R>(app: &mut S, req: R) -> S::Response
where
S: Service<Request = R>,
S::Error: std::fmt::Debug,
{
block_on(run_on(move || app.call(req))).unwrap()
}

2
actix-tower/CHANGES.md
View File

@ -1,2 +0,0 @@
# Changes

29
actix-tower/Cargo.toml
View File

@ -1,29 +0,0 @@
[package]
name = "actix-tower"
version = "0.1.0"
authors = ["Nikolay Kim <fafhrd91@gmail.com>", "Marcus Griep <marcus@griep.us>"]
description = "Actix Tower"
keywords = ["network", "framework", "async", "futures"]
homepage = "https://actix.rs"
repository = "https://github.com/actix/actix-net.git"
documentation = "https://docs.rs/actix-tower/"
categories = ["network-programming", "asynchronous"]
license = "MIT/Apache-2.0"
exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
edition = "2018"
workspace = ".."
[badges]
travis-ci = { repository = "actix/actix-tower", branch = "master" }
appveyor = { repository = "actix/actix-net" }
codecov = { repository = "actix/actix-tower", branch = "master", service = "github" }
[lib]
name = "actix_tower"
path = "src/lib.rs"
[dependencies]
actix-service = "0.3.6"
futures = "0.1.24"
tower-service = "0.2.0"

580
actix-tower/src/lib.rs
View File

@ -1,580 +0,0 @@
//! Utilities to provide interoperability between services based on the
//! `actix-service` and `tower-service` crates.
//!
//! ## Example
//!
//! In the following example, we take a `RandomService`—which will always
//! return 4—and wraps it with a middleware that will always add 1 to the
//! result. This pattern can be further used to wrap services from either
//! `tower-service` or `actix-service` with middleware provided by the other.
//!
//! ```
//! use actix_tower::ActixServiceExt;
//! # use futures::{Async, Future};
//! use actix_service::Service;
//!
//! struct RandomService;
//! impl Service for RandomService {
//! // …
//! # type Request = ();
//! # type Response = u32;
//! # type Error = ();
//! # type Future = futures::future::FutureResult<Self::Response, Self::Error>;
//! #
//! # fn poll_ready(&mut self) -> futures::Poll<(), Self::Error> {
//! # Ok(Async::Ready(()))
//! # }
//! #
//! # fn call(&mut self, _req: Self::Request) -> Self::Future {
//! # futures::finished(4)
//! # }
//! }
//!
//! struct AddOneMiddleware<S>(S);
//! impl<S, R> tower_service::Service<R> for AddOneMiddleware<S>
//! where
//! S: tower_service::Service<R, Response = u32>,
//! S::Future: 'static,
//! {
//! /// …
//! # type Response = u32;
//! # type Error = S::Error;
//! # type Future = Box<dyn Future<Item = Self::Response, Error = Self::Error>>;
//! #
//! # fn poll_ready(&mut self) -> futures::Poll<(), Self::Error> {
//! # self.0.poll_ready()
//! # }
//! #
//! # fn call(&mut self, req: R) -> Self::Future {
//! # let fut = self.0.call(req).map(|x| x + 1);
//! # Box::new(fut)
//! # }
//! }
//!
//! let mut s = RandomService.wrap_with_tower_middleware(AddOneMiddleware);
//! assert_eq!(Ok(Async::Ready(())), s.poll_ready());
//! assert_eq!(Ok(Async::Ready(5)), s.call(()).poll());
//! ```
use actix_service::Service as ActixService;
use std::marker::PhantomData;
use tower_service::Service as TowerService;
/// Compatibility wrapper associating a `tower_service::Service` with a particular
/// `Request` type, so that it can be used as an `actix_service::Service`.
///
/// Generally created through convenience methods on the `TowerServiceExt<R>` trait.
pub struct ActixCompat<S, R> {
inner: S,
_phantom: PhantomData<R>,
}
impl<S, R> ActixCompat<S, R> {
/// Wraps a `tower_service::Service` in a compatibility wrapper.
pub fn new(inner: S) -> Self {
ActixCompat {
inner,
_phantom: PhantomData,
}
}
}
/// Extension trait for wrapping a `tower_service::Service` instance for use as
/// an `actix_service::Service`.
pub trait TowerServiceExt<R>: TowerService<R> + Sized {
/// Wraps a `tower_service::Service` in a compatibility wrapper.
///
/// ```
/// use actix_service::Service;
/// use actix_tower::TowerServiceExt;
/// # use futures::{Async, Future};
///
/// struct RandomService;
/// impl<R> tower_service::Service<R> for RandomService {
/// // …
/// # type Response = u32;
/// # type Error = ();
/// # type Future = futures::future::FutureResult<Self::Response, Self::Error>;
/// #
/// # fn poll_ready(&mut self) -> futures::Poll<(), Self::Error> {
/// # Ok(Async::Ready(()))
/// # }
/// #
/// # fn call(&mut self, _req: R) -> Self::Future {
/// # futures::finished(4)
/// # }
/// }
///
/// let mut s = RandomService.into_actix_service();
/// assert_eq!(Ok(Async::Ready(())), s.poll_ready());
/// assert_eq!(Ok(Async::Ready(4)), s.call(()).poll());
/// ```
fn into_actix_service(self) -> ActixCompat<Self, R> {
ActixCompat::new(self)
}
/// Takes a function that, when provided with an `actix_service::Service` wraps it
/// and returns a new service. Useful for wrapping a `tower_service::Service` with
/// middleware built for `actix_service`.
///
/// ```
/// use actix_tower::TowerServiceExt;
/// # use futures::{Async, Future};
/// use tower_service::Service;
///
/// struct RandomService;
/// impl<R> Service<R> for RandomService {
/// // …
/// # type Response = u32;
/// # type Error = ();
/// # type Future = futures::future::FutureResult<Self::Response, Self::Error>;
/// #
/// # fn poll_ready(&mut self) -> futures::Poll<(), Self::Error> {
/// # Ok(Async::Ready(()))
/// # }
/// #
/// # fn call(&mut self, _req: R) -> Self::Future {
/// # futures::finished(4)
/// # }
/// }
///
/// struct AddOneTransform<S>(S);
/// impl<S> actix_service::Service for AddOneTransform<S>
/// where
/// S: actix_service::Service<Response = u32>,
/// S::Future: 'static,
/// {
/// /// …
/// # type Request = S::Request;
/// # type Response = u32;
/// # type Error = S::Error;
/// # type Future = Box<dyn Future<Item = Self::Response, Error = Self::Error>>;
/// #
/// # fn poll_ready(&mut self) -> futures::Poll<(), Self::Error> {
/// # self.0.poll_ready()
/// # }
/// #
/// # fn call(&mut self, req: Self::Request) -> Self::Future {
/// # let fut = self.0.call(req).map(|x| x + 1);
/// # Box::new(fut)
/// # }
/// }
///
/// let mut s = RandomService.wrap_with_actix_middleware(AddOneTransform);
/// assert_eq!(Ok(Async::Ready(())), s.poll_ready());
/// assert_eq!(Ok(Async::Ready(5)), s.call(()).poll());
/// ```
fn wrap_with_actix_middleware<F, U>(self, f: F) -> TowerCompat<U>
where
F: FnOnce(ActixCompat<Self, R>) -> U,
U: ActixService<Request = R>,
{
f(self.into_actix_service()).into_tower_service()
}
}
impl<S, R> TowerServiceExt<R> for S where S: TowerService<R> + Sized {}
impl<S, R> ActixService for ActixCompat<S, R>
where
S: TowerService<R>,
{
type Request = R;
type Response = S::Response;
type Error = S::Error;
type Future = S::Future;
fn poll_ready(&mut self) -> futures::Poll<(), Self::Error> {
TowerService::poll_ready(&mut self.inner)
}
fn call(&mut self, req: Self::Request) -> Self::Future {
TowerService::call(&mut self.inner, req)
}
}
/// Compatibility wrapper associating an `actix_service::Service` with a particular
/// `Request` type, so that it can be used as a `tower_service::Service`.
///
/// Generally created through convenience methods on the `ActixServiceExt<R>` trait.
pub struct TowerCompat<S> {
inner: S,
}
impl<S> TowerCompat<S> {
/// Wraps an `actix_service::Service` in a compatibility wrapper.
pub fn new(inner: S) -> Self {
TowerCompat { inner }
}
}
/// Extension trait for wrapping an `actix_service::Service` instance for use as
/// a `tower_service::Service`.
pub trait ActixServiceExt: ActixService + Sized {
/// Wraps a `tower_service::Service` in a compatibility wrapper.
///
/// ```
/// use actix_tower::ActixServiceExt;
/// # use futures::{Async, Future};
/// use tower_service::Service;
///
/// struct RandomService;
/// impl actix_service::Service for RandomService {
/// // …
/// # type Request = ();
/// # type Response = u32;
/// # type Error = ();
/// # type Future = futures::future::FutureResult<Self::Response, Self::Error>;
/// #
/// # fn poll_ready(&mut self) -> futures::Poll<(), Self::Error> {
/// # Ok(Async::Ready(()))
/// # }
/// #
/// # fn call(&mut self, _req: Self::Request) -> Self::Future {
/// # futures::finished(4)
/// # }
/// }
///
/// let mut s = RandomService.into_tower_service();
/// assert_eq!(Ok(Async::Ready(())), s.poll_ready());
/// assert_eq!(Ok(Async::Ready(4)), s.call(()).poll());
/// ```
fn into_tower_service(self) -> TowerCompat<Self> {
TowerCompat::new(self)
}
/// Takes a function that, when provided with a `tower_service::Service` wraps it
/// and returns a new service. Useful for wrapping an `actix_service::Service` with
/// middleware built for `tower_service`.
///
/// ```
/// use actix_tower::ActixServiceExt;
/// # use futures::{Async, Future};
/// use actix_service::Service;
///
/// struct RandomService;
/// impl Service for RandomService {
/// // …
/// # type Request = ();
/// # type Response = u32;
/// # type Error = ();
/// # type Future = futures::future::FutureResult<Self::Response, Self::Error>;
/// #
/// # fn poll_ready(&mut self) -> futures::Poll<(), Self::Error> {
/// # Ok(Async::Ready(()))
/// # }
/// #
/// # fn call(&mut self, _req: Self::Request) -> Self::Future {
/// # futures::finished(4)
/// # }
/// }
///
/// struct AddOneMiddleware<S>(S);
/// impl<S, R> tower_service::Service<R> for AddOneMiddleware<S>
/// where
/// S: tower_service::Service<R, Response = u32>,
/// S::Future: 'static,
/// {
/// /// …
/// # type Response = u32;
/// # type Error = S::Error;
/// # type Future = Box<dyn Future<Item = Self::Response, Error = Self::Error>>;
/// #
/// # fn poll_ready(&mut self) -> futures::Poll<(), Self::Error> {
/// # self.0.poll_ready()
/// # }
/// #
/// # fn call(&mut self, req: R) -> Self::Future {
/// # let fut = self.0.call(req).map(|x| x + 1);
/// # Box::new(fut)
/// # }
/// }
///
/// let mut s = RandomService.wrap_with_tower_middleware(AddOneMiddleware);
/// assert_eq!(Ok(Async::Ready(())), s.poll_ready());
/// assert_eq!(Ok(Async::Ready(5)), s.call(()).poll());
/// ```
fn wrap_with_tower_middleware<F, U>(self, f: F) -> ActixCompat<U, Self::Request>
where
F: FnOnce(TowerCompat<Self>) -> U,
U: TowerService<Self::Request>,
{
f(self.into_tower_service()).into_actix_service()
}
}
impl<S> ActixServiceExt for S where S: ActixService + Sized {}
impl<S> TowerService<S::Request> for TowerCompat<S>
where
S: ActixService,
{
type Response = S::Response;
type Error = S::Error;
type Future = S::Future;
fn poll_ready(&mut self) -> futures::Poll<(), Self::Error> {
ActixService::poll_ready(&mut self.inner)
}
fn call(&mut self, req: S::Request) -> Self::Future {
ActixService::call(&mut self.inner, req)
}
}
#[cfg(test)]
mod tests {
mod tower_service_into_actix_service {
use crate::TowerServiceExt;
use actix_service::{Service as ActixService, ServiceExt, Transform};
use futures::{future::FutureResult, Async, Future, Poll};
use tower_service::Service as TowerService;
#[test]
fn random_service_returns_4() {
let mut s = RandomService.into_actix_service();
assert_eq!(Ok(Async::Ready(())), s.poll_ready());
assert_eq!(Ok(Async::Ready(4)), s.call(()).poll());
}
#[test]
fn random_service_can_combine() {
let mut s = RandomService.into_actix_service().map(|x| x + 1);
assert_eq!(Ok(Async::Ready(())), s.poll_ready());
assert_eq!(Ok(Async::Ready(5)), s.call(()).poll());
}
#[test]
fn random_service_can_use_actix_middleware() {
let mut s = RandomService.wrap_with_actix_middleware(DoMathTransform);
assert_eq!(Ok(Async::Ready(())), s.poll_ready());
assert_eq!(Ok(Async::Ready(68)), s.call(()).poll());
}
#[test]
fn random_service_and_add_service_chained() {
let s1 = RandomService.into_actix_service();
let s2 = AddOneService.into_actix_service();
let s3 = AddOneService.into_actix_service();
let mut s = s1.and_then(s2).and_then(s3);
assert_eq!(Ok(Async::Ready(())), s.poll_ready());
assert_eq!(Ok(Async::Ready(6)), s.call(()).poll());
}
#[test]
fn random_service_and_add_service_and_ignoring_service_chained() {
let s1 = RandomService.into_actix_service();
let s2 = AddOneService.into_actix_service();
let s3 = AddOneService.into_actix_service();
let s4 = RandomService.into_actix_service();
let mut s = s1.and_then(s2).and_then(s3).and_then(s4);
assert_eq!(Ok(Async::Ready(())), s.poll_ready());
assert_eq!(Ok(Async::Ready(4)), s.call(()).poll());
}
#[test]
fn random_service_can_be_transformed_to_do_math() {
let transform = DoMath;
let mut s = transform
.new_transform(RandomService.into_actix_service())
.wait()
.unwrap();
assert_eq!(Ok(Async::Ready(())), s.poll_ready());
assert_eq!(Ok(Async::Ready(68)), s.call(()).poll());
}
struct RandomService;
impl<R> TowerService<R> for RandomService {
type Response = u32;
type Error = ();
type Future = FutureResult<Self::Response, Self::Error>;
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
Ok(Async::Ready(()))
}
fn call(&mut self, _req: R) -> Self::Future {
futures::finished(4)
}
}
struct AddOneService;
impl TowerService<u32> for AddOneService {
type Response = u32;
type Error = ();
type Future = FutureResult<Self::Response, Self::Error>;
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
Ok(Async::Ready(()))
}
fn call(&mut self, req: u32) -> Self::Future {
futures::finished(req + 1)
}
}
struct DoMathTransform<S>(S);
impl<S> ActixService for DoMathTransform<S>
where
S: ActixService<Response = u32>,
S::Future: 'static,
{
type Request = S::Request;
type Response = u32;
type Error = S::Error;
type Future = Box<dyn Future<Item = Self::Response, Error = Self::Error>>;
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
self.0.poll_ready()
}
fn call(&mut self, req: Self::Request) -> Self::Future {
let fut = self.0.call(req).map(|x| x * 17);
Box::new(fut)
}
}
struct DoMath;
impl<S> Transform<S> for DoMath
where
S: ActixService<Response = u32>,
S::Future: 'static,
{
type Request = S::Request;
type Response = u32;
type Error = S::Error;
type Transform = DoMathTransform<S>;
type InitError = ();
type Future = FutureResult<Self::Transform, Self::InitError>;
fn new_transform(&self, service: S) -> Self::Future {
futures::finished(DoMathTransform(service))
}
}
}
mod actix_service_into_tower_service {
use crate::{ActixServiceExt, TowerServiceExt};
use actix_service::{Service as ActixService, ServiceExt};
use futures::{future::FutureResult, Async, Future, Poll};
use tower_service::Service as TowerService;
#[test]
fn random_service_returns_4() {
let mut s = RandomService.into_tower_service();
assert_eq!(Ok(Async::Ready(())), s.poll_ready());
assert_eq!(Ok(Async::Ready(4)), s.call(()).poll());
}
#[test]
fn random_service_can_use_tower_middleware() {
let mut s =
AddOneService::wrap(RandomService.into_tower_service()).into_actix_service();
assert_eq!(Ok(Async::Ready(())), s.poll_ready());
assert_eq!(Ok(Async::Ready(5)), s.call(()).poll());
}
#[test]
fn do_math_service_can_use_tower_middleware() {
let mut s =
AddOneService::wrap(DoMathService.into_tower_service()).into_actix_service();
assert_eq!(Ok(Async::Ready(())), s.poll_ready());
assert_eq!(Ok(Async::Ready(188)), s.call(11).poll());
}
#[test]
fn random_service_and_add_service_and_ignoring_service_chained() {
let s1 = RandomService.wrap_with_tower_middleware(AddOneService::wrap);
let s2 = DoMathService.wrap_with_tower_middleware(AddOneService::wrap);
let mut s = s1.and_then(s2);
assert_eq!(Ok(Async::Ready(())), s.poll_ready());
assert_eq!(Ok(Async::Ready(86)), s.call(()).poll());
}
struct RandomService;
impl ActixService for RandomService {
type Request = ();
type Response = u32;
type Error = ();
type Future = FutureResult<Self::Response, Self::Error>;
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
Ok(Async::Ready(()))
}
fn call(&mut self, _req: Self::Request) -> Self::Future {
futures::finished(4)
}
}
struct AddOneService<S> {
inner: S,
}
impl<S> AddOneService<S> {
fn wrap(inner: S) -> Self {
AddOneService { inner }
}
}
impl<S, R> TowerService<R> for AddOneService<S>
where
S: TowerService<R, Response = u32>,
S::Future: 'static,
{
type Response = u32;
type Error = S::Error;
type Future = Box<dyn Future<Item = Self::Response, Error = Self::Error>>;
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
self.inner.poll_ready()
}
fn call(&mut self, req: R) -> Self::Future {
let fut = self.inner.call(req).map(|x| x + 1);
Box::new(fut)
}
}
struct DoMathService;
impl ActixService for DoMathService {
type Request = u32;
type Response = u32;
type Error = ();
type Future = FutureResult<Self::Response, Self::Error>;
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
Ok(Async::Ready(()))
}
fn call(&mut self, req: Self::Request) -> Self::Future {
futures::finished(req * 17)
}
}
}
}