Merge branch 'master' of https://github.com/actix/actix-net into unsafe-cell

This commit is contained in:
Maxim Vorobjov 2020-06-26 18:48:30 +03:00
commit a37c3fb00b
No known key found for this signature in database
GPG Key ID: 3A6CF6ADF8859751
81 changed files with 515 additions and 293 deletions

View File

@ -29,18 +29,16 @@ jobs:
uses: actions-rs/cargo@v1 uses: actions-rs/cargo@v1
with: with:
command: generate-lockfile command: generate-lockfile
- name: Cache cargo registry - name: Cache cargo dirs
uses: actions/cache@v1 uses: actions/cache@v2
with: with:
path: ~/.cargo/registry path:
key: ${{ matrix.version }}-x86_64-unknown-linux-gnu-cargo-registry-trimmed-${{ hashFiles('**/Cargo.lock') }} ~/.cargo/registry
- name: Cache cargo index ~/.cargo/git
uses: actions/cache@v1 ~/.cargo/bin
with: key: ${{ matrix.version }}-x86_64-unknown-linux-gnu-cargo-trimmed-${{ hashFiles('**/Cargo.lock') }}
path: ~/.cargo/git
key: ${{ matrix.version }}-x86_64-unknown-linux-gnu-cargo-index-trimmed-${{ hashFiles('**/Cargo.lock') }}
- name: Cache cargo build - name: Cache cargo build
uses: actions/cache@v1 uses: actions/cache@v2
with: with:
path: target path: target
key: ${{ matrix.version }}-x86_64-unknown-linux-gnu-cargo-build-trimmed-${{ hashFiles('**/Cargo.lock') }} key: ${{ matrix.version }}-x86_64-unknown-linux-gnu-cargo-build-trimmed-${{ hashFiles('**/Cargo.lock') }}
@ -59,19 +57,18 @@ jobs:
args: --all --all-features --no-fail-fast -- --nocapture args: --all --all-features --no-fail-fast -- --nocapture
- name: Generate coverage file - name: Generate coverage file
if: matrix.version == 'stable' && github.ref == 'refs/heads/master' if: matrix.version == 'stable' && (github.ref == 'refs/heads/master' || github.event_name == 'pull_request')
run: | run: |
cargo install cargo-tarpaulin which cargo-tarpaulin || cargo install cargo-tarpaulin
cargo tarpaulin --out Xml --workspace --all-features cargo tarpaulin --out Xml --workspace --all-features
- name: Upload to Codecov - name: Upload to Codecov
if: matrix.version == 'stable' && github.ref == 'refs/heads/master' if: matrix.version == 'stable' && (github.ref == 'refs/heads/master' || github.event_name == 'pull_request')
uses: codecov/codecov-action@v1 uses: codecov/codecov-action@v1
with: with:
token: ${{ secrets.CODECOV_TOKEN }}
file: cobertura.xml file: cobertura.xml
- name: Clear the cargo caches - name: Clear the cargo caches
run: | run: |
cargo install cargo-cache --no-default-features --features ci-autoclean which cargo-cache || cargo install cargo-cache --no-default-features --features ci-autoclean
cargo-cache cargo-cache

View File

@ -2,9 +2,6 @@ name: CI (Windows-mingw)
on: [push, pull_request] on: [push, pull_request]
env:
OPENSSL_DIR: d:\a\_temp\msys\msys64\usr
jobs: jobs:
build_and_test: build_and_test:
strategy: strategy:
@ -30,25 +27,13 @@ jobs:
- name: Install MSYS2 - name: Install MSYS2
uses: numworks/setup-msys2@v1 uses: numworks/setup-msys2@v1
- name: Install OpenSSL - name: Install packages
run: | run: |
msys2do pacman --noconfirm -S openssl-devel pkg-config msys2do pacman -Sy --noconfirm pacman
msys2do pacman --noconfirm -S base-devel pkg-config
- name: Copy and check libs
run: |
Copy-Item d:\a\_temp\msys\msys64\usr\lib\libssl.dll.a d:\a\_temp\msys\msys64\usr\lib\libssl.dll
Copy-Item d:\a\_temp\msys\msys64\usr\lib\libcrypto.dll.a d:\a\_temp\msys\msys64\usr\lib\libcrypto.dll
Get-ChildItem d:\a\_temp\msys\msys64\usr\lib
Get-ChildItem d:\a\_temp\msys\msys64\usr
- name: check build - name: check build
uses: actions-rs/cargo@v1 uses: actions-rs/cargo@v1
with: with:
command: check command: check
args: --all --bins --examples --tests args: --all --bins --examples --tests
- name: tests
uses: actions-rs/cargo@v1
with:
command: test
args: --all --all-features --no-fail-fast -- --nocapture

View File

@ -19,9 +19,9 @@ path = "src/lib.rs"
[dependencies] [dependencies]
bitflags = "1.2.1" bitflags = "1.2.1"
bytes = "0.5.2" bytes = "0.5.2"
futures-core = "0.3.1" futures-core = { version = "0.3.4", default-features = false }
futures-sink = "0.3.1" futures-sink = { version = "0.3.4", default-features = false }
tokio = { version = "0.2.4", default-features=false } tokio = { version = "0.2.4", default-features=false }
tokio-util = { version = "0.2.0", default-features=false, features=["codec"] } tokio-util = { version = "0.2.0", default-features=false, features=["codec"] }
log = "0.4" log = "0.4"
pin-project = "0.4.8" pin-project = "0.4.17"

View File

@ -2,10 +2,12 @@
//! //!
//! Contains adapters to go from streams of bytes, [`AsyncRead`] and //! Contains adapters to go from streams of bytes, [`AsyncRead`] and
//! [`AsyncWrite`], to framed streams implementing [`Sink`] and [`Stream`]. //! [`AsyncWrite`], to framed streams implementing [`Sink`] and [`Stream`].
//! Framed streams are also known as [transports]. //! Framed streams are also known as `transports`.
//! //!
//! [`AsyncRead`]: # //! [`AsyncRead`]: AsyncRead
//! [`AsyncWrite`]: # //! [`AsyncWrite`]: AsyncWrite
//! [`Sink`]: futures_sink::Sink
//! [`Stream`]: futures_core::Stream
#![deny(rust_2018_idioms, warnings)] #![deny(rust_2018_idioms, warnings)]
mod bcodec; mod bcodec;

View File

@ -1,5 +1,22 @@
# Changes # Changes
## [2.0.0-alpha.3] - 2020-05-08
### Fixed
* Corrected spelling of `ConnectError::Unresolverd` to `ConnectError::Unresolved`
## [2.0.0-alpha.2] - 2020-03-08
### Changed
* Update `trust-dns-proto` dependency to 0.19. [#116]
* Update `trust-dns-resolver` dependency to 0.19. [#116]
* `Address` trait is now required to have static lifetime. [#116]
* `start_resolver` and `start_default_resolver` are now `async` and may return a `ConnectError`. [#116]
[#116]: https://github.com/actix/actix-net/pull/116
## [2.0.0-alpha.1] - 2020-03-03 ## [2.0.0-alpha.1] - 2020-03-03
### Changed ### Changed

View File

@ -1,6 +1,6 @@
[package] [package]
name = "actix-connect" name = "actix-connect"
version = "2.0.0-alpha.1" version = "2.0.0-alpha.3"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"] authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix connect - tcp connector service" description = "Actix connect - tcp connector service"
keywords = ["network", "framework", "async", "futures"] keywords = ["network", "framework", "async", "futures"]
@ -37,11 +37,11 @@ actix-utils = "1.0.6"
actix-rt = "1.0.0" actix-rt = "1.0.0"
derive_more = "0.99.2" derive_more = "0.99.2"
either = "1.5.3" either = "1.5.3"
futures = "0.3.1" futures-util = { version = "0.3.4", default-features = false }
http = { version = "0.2.0", optional = true } http = { version = "0.2.0", optional = true }
log = "0.4" log = "0.4"
trust-dns-proto = "=0.18.0-alpha.2" trust-dns-proto = { version = "0.19", default-features = false, features = ["tokio-runtime"] }
trust-dns-resolver = "=0.18.0-alpha.2" trust-dns-resolver = { version = "0.19", default-features = false, features = ["tokio-runtime", "system-config"] }
# openssl # openssl
open-ssl = { version="0.10", package = "openssl", optional = true } open-ssl = { version="0.10", package = "openssl", optional = true }

View File

@ -6,7 +6,7 @@ use std::net::SocketAddr;
use either::Either; use either::Either;
/// Connect request /// Connect request
pub trait Address: Unpin { pub trait Address: Unpin + 'static {
/// Host name of the request /// Host name of the request
fn host(&self) -> &str; fn host(&self) -> &str;

View File

@ -8,7 +8,7 @@ use std::task::{Context, Poll};
use actix_rt::net::TcpStream; use actix_rt::net::TcpStream;
use actix_service::{Service, ServiceFactory}; use actix_service::{Service, ServiceFactory};
use futures::future::{err, ok, BoxFuture, Either, FutureExt, Ready}; use futures_util::future::{err, ok, BoxFuture, Either, FutureExt, Ready};
use super::connect::{Address, Connect, Connection}; use super::connect::{Address, Connect, Connection};
use super::error::ConnectError; use super::error::ConnectError;
@ -88,7 +88,7 @@ impl<T: Address> Service for TcpConnector<T> {
Either::Left(TcpConnectorResponse::new(req, port, addr)) Either::Left(TcpConnectorResponse::new(req, port, addr))
} else { } else {
error!("TCP connector: got unresolved address"); error!("TCP connector: got unresolved address");
Either::Right(err(ConnectError::Unresolverd)) Either::Right(err(ConnectError::Unresolved))
} }
} }
} }

View File

@ -18,7 +18,7 @@ pub enum ConnectError {
/// Unresolved host name /// Unresolved host name
#[display(fmt = "Connector received `Connect` method with unresolved host")] #[display(fmt = "Connector received `Connect` method with unresolved host")]
Unresolverd, Unresolved,
/// Connection io error /// Connection io error
#[display(fmt = "{}", _0)] #[display(fmt = "{}", _0)]

View File

@ -25,7 +25,7 @@ use actix_rt::{net::TcpStream, Arbiter};
use actix_service::{pipeline, pipeline_factory, Service, ServiceFactory}; use actix_service::{pipeline, pipeline_factory, Service, ServiceFactory};
use trust_dns_resolver::config::{ResolverConfig, ResolverOpts}; use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
use trust_dns_resolver::system_conf::read_system_conf; use trust_dns_resolver::system_conf::read_system_conf;
use trust_dns_resolver::AsyncResolver; use trust_dns_resolver::TokioAsyncResolver as AsyncResolver;
pub mod resolver { pub mod resolver {
pub use trust_dns_resolver::config::{ResolverConfig, ResolverOpts}; pub use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
@ -39,17 +39,18 @@ pub use self::error::ConnectError;
pub use self::resolve::{Resolver, ResolverFactory}; pub use self::resolve::{Resolver, ResolverFactory};
pub use self::service::{ConnectService, ConnectServiceFactory, TcpConnectService}; pub use self::service::{ConnectService, ConnectServiceFactory, TcpConnectService};
pub fn start_resolver(cfg: ResolverConfig, opts: ResolverOpts) -> AsyncResolver { pub async fn start_resolver(
let (resolver, bg) = AsyncResolver::new(cfg, opts); cfg: ResolverConfig,
actix_rt::spawn(bg); opts: ResolverOpts,
resolver ) -> Result<AsyncResolver, ConnectError> {
Ok(AsyncResolver::tokio(cfg, opts).await?)
} }
struct DefaultResolver(AsyncResolver); struct DefaultResolver(AsyncResolver);
pub(crate) fn get_default_resolver() -> AsyncResolver { pub(crate) async fn get_default_resolver() -> Result<AsyncResolver, ConnectError> {
if Arbiter::contains_item::<DefaultResolver>() { if Arbiter::contains_item::<DefaultResolver>() {
Arbiter::get_item(|item: &DefaultResolver| item.0.clone()) Ok(Arbiter::get_item(|item: &DefaultResolver| item.0.clone()))
} else { } else {
let (cfg, opts) = match read_system_conf() { let (cfg, opts) = match read_system_conf() {
Ok((cfg, opts)) => (cfg, opts), Ok((cfg, opts)) => (cfg, opts),
@ -59,16 +60,15 @@ pub(crate) fn get_default_resolver() -> AsyncResolver {
} }
}; };
let (resolver, bg) = AsyncResolver::new(cfg, opts); let resolver = AsyncResolver::tokio(cfg, opts).await?;
actix_rt::spawn(bg);
Arbiter::set_item(DefaultResolver(resolver.clone())); Arbiter::set_item(DefaultResolver(resolver.clone()));
resolver Ok(resolver)
} }
} }
pub fn start_default_resolver() -> AsyncResolver { pub async fn start_default_resolver() -> Result<AsyncResolver, ConnectError> {
get_default_resolver() get_default_resolver().await
} }
/// Create tcp connector service /// Create tcp connector service

View File

@ -5,9 +5,9 @@ use std::pin::Pin;
use std::task::{Context, Poll}; use std::task::{Context, Poll};
use actix_service::{Service, ServiceFactory}; use actix_service::{Service, ServiceFactory};
use futures::future::{ok, Either, Ready}; use futures_util::future::{ok, Either, Ready};
use trust_dns_resolver::lookup_ip::LookupIpFuture; use trust_dns_resolver::TokioAsyncResolver as AsyncResolver;
use trust_dns_resolver::{AsyncResolver, Background}; use trust_dns_resolver::{error::ResolveError, lookup_ip::LookupIp};
use crate::connect::{Address, Connect}; use crate::connect::{Address, Connect};
use crate::error::ConnectError; use crate::error::ConnectError;
@ -106,7 +106,10 @@ impl<T: Address> Service for Resolver<T> {
type Request = Connect<T>; type Request = Connect<T>;
type Response = Connect<T>; type Response = Connect<T>;
type Error = ConnectError; type Error = ConnectError;
type Future = Either<ResolverFuture<T>, Ready<Result<Connect<T>, Self::Error>>>; type Future = Either<
Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>,
Ready<Result<Connect<T>, Self::Error>>,
>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> { fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(())) Poll::Ready(Ok(()))
@ -119,32 +122,48 @@ impl<T: Address> Service for Resolver<T> {
req.addr = Some(either::Either::Left(SocketAddr::new(ip, req.port()))); req.addr = Some(either::Either::Left(SocketAddr::new(ip, req.port())));
Either::Right(ok(req)) Either::Right(ok(req))
} else { } else {
trace!("DNS resolver: resolving host {:?}", req.host()); let resolver = self.resolver.as_ref().map(AsyncResolver::clone);
if self.resolver.is_none() { Either::Left(Box::pin(async move {
self.resolver = Some(get_default_resolver()); trace!("DNS resolver: resolving host {:?}", req.host());
} let resolver = if let Some(resolver) = resolver {
Either::Left(ResolverFuture::new(req, self.resolver.as_ref().unwrap())) resolver
} else {
get_default_resolver()
.await
.expect("Failed to get default resolver")
};
ResolverFuture::new(req, &resolver).await
}))
} }
} }
} }
type LookupIpFuture = Pin<Box<dyn Future<Output = Result<LookupIp, ResolveError>>>>;
#[doc(hidden)] #[doc(hidden)]
/// Resolver future /// Resolver future
pub struct ResolverFuture<T: Address> { pub struct ResolverFuture<T: Address> {
req: Option<Connect<T>>, req: Option<Connect<T>>,
lookup: Background<LookupIpFuture>, lookup: LookupIpFuture,
} }
impl<T: Address> ResolverFuture<T> { impl<T: Address> ResolverFuture<T> {
pub fn new(req: Connect<T>, resolver: &AsyncResolver) -> Self { pub fn new(req: Connect<T>, resolver: &AsyncResolver) -> Self {
let lookup = if let Some(host) = req.host().splitn(2, ':').next() { let host = if let Some(host) = req.host().splitn(2, ':').next() {
resolver.lookup_ip(host) host
} else { } else {
resolver.lookup_ip(req.host()) req.host()
}; };
// Clone data to be moved to the lookup future
let host_clone = host.to_owned();
let resolver_clone = resolver.clone();
ResolverFuture { ResolverFuture {
lookup, lookup: Box::pin(async move {
let resolver = resolver_clone;
resolver.lookup_ip(host_clone).await
}),
req: Some(req), req: Some(req),
} }
} }

View File

@ -5,8 +5,8 @@ use std::task::{Context, Poll};
use actix_rt::net::TcpStream; use actix_rt::net::TcpStream;
use actix_service::{Service, ServiceFactory}; use actix_service::{Service, ServiceFactory};
use either::Either; use either::Either;
use futures::future::{ok, Ready}; use futures_util::future::{ok, Ready};
use trust_dns_resolver::AsyncResolver; use trust_dns_resolver::TokioAsyncResolver as AsyncResolver;
use crate::connect::{Address, Connect, Connection}; use crate::connect::{Address, Connect, Connection};
use crate::connector::{TcpConnector, TcpConnectorFactory}; use crate::connector::{TcpConnector, TcpConnectorFactory};

View File

@ -10,8 +10,8 @@ pub use tokio_openssl::{HandshakeError, SslStream};
use actix_codec::{AsyncRead, AsyncWrite}; use actix_codec::{AsyncRead, AsyncWrite};
use actix_rt::net::TcpStream; use actix_rt::net::TcpStream;
use actix_service::{Service, ServiceFactory}; use actix_service::{Service, ServiceFactory};
use futures::future::{err, ok, Either, FutureExt, LocalBoxFuture, Ready}; use futures_util::future::{err, ok, Either, FutureExt, LocalBoxFuture, Ready};
use trust_dns_resolver::AsyncResolver; use trust_dns_resolver::TokioAsyncResolver as AsyncResolver;
use crate::{ use crate::{
Address, Connect, ConnectError, ConnectService, ConnectServiceFactory, Connection, Address, Connect, ConnectError, ConnectService, ConnectServiceFactory, Connection,
@ -243,7 +243,7 @@ impl<T: Address> Future for OpensslConnectServiceResponse<T> {
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
if let Some(ref mut fut) = self.fut1 { if let Some(ref mut fut) = self.fut1 {
match futures::ready!(Pin::new(fut).poll(cx)) { match futures_util::ready!(Pin::new(fut).poll(cx)) {
Ok(res) => { Ok(res) => {
let _ = self.fut1.take(); let _ = self.fut1.take();
self.fut2 = Some(self.openssl.call(res)); self.fut2 = Some(self.openssl.call(res));
@ -253,7 +253,7 @@ impl<T: Address> Future for OpensslConnectServiceResponse<T> {
} }
if let Some(ref mut fut) = self.fut2 { if let Some(ref mut fut) = self.fut2 {
match futures::ready!(Pin::new(fut).poll(cx)) { match futures_util::ready!(Pin::new(fut).poll(cx)) {
Ok(connect) => Poll::Ready(Ok(connect.into_parts().0)), Ok(connect) => Poll::Ready(Ok(connect.into_parts().0)),
Err(e) => Poll::Ready(Err(ConnectError::Io(io::Error::new( Err(e) => Poll::Ready(Err(ConnectError::Io(io::Error::new(
io::ErrorKind::Other, io::ErrorKind::Other,

View File

@ -10,7 +10,7 @@ pub use tokio_rustls::{client::TlsStream, rustls::ClientConfig};
use actix_codec::{AsyncRead, AsyncWrite}; use actix_codec::{AsyncRead, AsyncWrite};
use actix_service::{Service, ServiceFactory}; use actix_service::{Service, ServiceFactory};
use futures::future::{ok, Ready}; use futures_util::future::{ok, Ready};
use tokio_rustls::{Connect, TlsConnector}; use tokio_rustls::{Connect, TlsConnector};
use webpki::DNSNameRef; use webpki::DNSNameRef;
@ -126,7 +126,7 @@ where
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.get_mut(); let this = self.get_mut();
Poll::Ready( Poll::Ready(
futures::ready!(Pin::new(&mut this.fut).poll(cx)).map(|stream| { futures_util::ready!(Pin::new(&mut this.fut).poll(cx)).map(|stream| {
let s = this.stream.take().unwrap(); let s = this.stream.take().unwrap();
trace!("SSL Handshake success: {:?}", s.host()); trace!("SSL Handshake success: {:?}", s.host());
s.replace(stream).1 s.replace(stream).1

View File

@ -5,7 +5,7 @@ use actix_rt::net::TcpStream;
use actix_service::{fn_service, Service, ServiceFactory}; use actix_service::{fn_service, Service, ServiceFactory};
use actix_testing::TestServer; use actix_testing::TestServer;
use bytes::Bytes; use bytes::Bytes;
use futures::SinkExt; use futures_util::sink::SinkExt;
use actix_connect::resolver::{ResolverConfig, ResolverOpts}; use actix_connect::resolver::{ResolverConfig, ResolverOpts};
use actix_connect::Connect; use actix_connect::Connect;
@ -14,12 +14,10 @@ use actix_connect::Connect;
#[actix_rt::test] #[actix_rt::test]
async fn test_string() { async fn test_string() {
let srv = TestServer::with(|| { let srv = TestServer::with(|| {
fn_service(|io: TcpStream| { fn_service(|io: TcpStream| async {
async { let mut framed = Framed::new(io, BytesCodec);
let mut framed = Framed::new(io, BytesCodec); framed.send(Bytes::from_static(b"test")).await?;
framed.send(Bytes::from_static(b"test")).await?; Ok::<_, io::Error>(())
Ok::<_, io::Error>(())
}
}) })
}); });
@ -33,12 +31,10 @@ async fn test_string() {
#[actix_rt::test] #[actix_rt::test]
async fn test_rustls_string() { async fn test_rustls_string() {
let srv = TestServer::with(|| { let srv = TestServer::with(|| {
fn_service(|io: TcpStream| { fn_service(|io: TcpStream| async {
async { let mut framed = Framed::new(io, BytesCodec);
let mut framed = Framed::new(io, BytesCodec); framed.send(Bytes::from_static(b"test")).await?;
framed.send(Bytes::from_static(b"test")).await?; Ok::<_, io::Error>(())
Ok::<_, io::Error>(())
}
}) })
}); });
@ -51,16 +47,14 @@ async fn test_rustls_string() {
#[actix_rt::test] #[actix_rt::test]
async fn test_static_str() { async fn test_static_str() {
let srv = TestServer::with(|| { let srv = TestServer::with(|| {
fn_service(|io: TcpStream| { fn_service(|io: TcpStream| async {
async { let mut framed = Framed::new(io, BytesCodec);
let mut framed = Framed::new(io, BytesCodec); framed.send(Bytes::from_static(b"test")).await?;
framed.send(Bytes::from_static(b"test")).await?; Ok::<_, io::Error>(())
Ok::<_, io::Error>(())
}
}) })
}); });
let resolver = actix_connect::start_default_resolver(); let resolver = actix_connect::start_default_resolver().await.unwrap();
let mut conn = actix_connect::new_connector(resolver.clone()); let mut conn = actix_connect::new_connector(resolver.clone());
let con = conn.call(Connect::with("10", srv.addr())).await.unwrap(); let con = conn.call(Connect::with("10", srv.addr())).await.unwrap();
@ -75,17 +69,17 @@ async fn test_static_str() {
#[actix_rt::test] #[actix_rt::test]
async fn test_new_service() { async fn test_new_service() {
let srv = TestServer::with(|| { let srv = TestServer::with(|| {
fn_service(|io: TcpStream| { fn_service(|io: TcpStream| async {
async { let mut framed = Framed::new(io, BytesCodec);
let mut framed = Framed::new(io, BytesCodec); framed.send(Bytes::from_static(b"test")).await?;
framed.send(Bytes::from_static(b"test")).await?; Ok::<_, io::Error>(())
Ok::<_, io::Error>(())
}
}) })
}); });
let resolver = let resolver =
actix_connect::start_resolver(ResolverConfig::default(), ResolverOpts::default()); actix_connect::start_resolver(ResolverConfig::default(), ResolverOpts::default())
.await
.unwrap();
let factory = actix_connect::new_connector_factory(resolver); let factory = actix_connect::new_connector_factory(resolver);
@ -100,12 +94,10 @@ async fn test_uri() {
use std::convert::TryFrom; use std::convert::TryFrom;
let srv = TestServer::with(|| { let srv = TestServer::with(|| {
fn_service(|io: TcpStream| { fn_service(|io: TcpStream| async {
async { let mut framed = Framed::new(io, BytesCodec);
let mut framed = Framed::new(io, BytesCodec); framed.send(Bytes::from_static(b"test")).await?;
framed.send(Bytes::from_static(b"test")).await?; Ok::<_, io::Error>(())
Ok::<_, io::Error>(())
}
}) })
}); });
@ -121,12 +113,10 @@ async fn test_rustls_uri() {
use std::convert::TryFrom; use std::convert::TryFrom;
let srv = TestServer::with(|| { let srv = TestServer::with(|| {
fn_service(|io: TcpStream| { fn_service(|io: TcpStream| async {
async { let mut framed = Framed::new(io, BytesCodec);
let mut framed = Framed::new(io, BytesCodec); framed.send(Bytes::from_static(b"test")).await?;
framed.send(Bytes::from_static(b"test")).await?; Ok::<_, io::Error>(())
Ok::<_, io::Error>(())
}
}) })
}); });

View File

@ -22,10 +22,12 @@ actix-utils = "1.0.4"
actix-rt = "1.0.0" actix-rt = "1.0.0"
bytes = "0.5.3" bytes = "0.5.3"
either = "1.5.3" either = "1.5.3"
futures = "0.3.1" futures-sink = { version = "0.3.4", default-features = false }
pin-project = "0.4.6" futures-core = { version = "0.3.4", default-features = false }
pin-project = "0.4.17"
log = "0.4" log = "0.4"
[dev-dependencies] [dev-dependencies]
actix-connect = "2.0.0-alpha.1" actix-connect = "2.0.0-alpha.2"
actix-testing = "1.0.0" actix-testing = "1.0.0"
futures-util = { version = "0.3.4", default-features = false }

View File

@ -4,7 +4,7 @@ use std::task::{Context, Poll};
use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed}; use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
use actix_utils::mpsc::Receiver; use actix_utils::mpsc::Receiver;
use futures::Stream; use futures_core::stream::Stream;
pub struct Connect<Io, Codec> pub struct Connect<Io, Codec>
where where
@ -90,7 +90,7 @@ where
} }
} }
impl<Io, St, Codec, Out> futures::Sink<<Codec as Encoder>::Item> impl<Io, St, Codec, Out> futures_sink::Sink<<Codec as Encoder>::Item>
for ConnectResult<Io, St, Codec, Out> for ConnectResult<Io, St, Codec, Out>
where where
Io: AsyncRead + AsyncWrite, Io: AsyncRead + AsyncWrite,

View File

@ -5,7 +5,7 @@ use std::task::{Context, Poll};
use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed}; use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
use actix_service::Service; use actix_service::Service;
use actix_utils::mpsc; use actix_utils::mpsc;
use futures::Stream; use futures_core::stream::Stream;
use pin_project::pin_project; use pin_project::pin_project;
use log::debug; use log::debug;

View File

@ -7,8 +7,7 @@ use std::task::{Context, Poll};
use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed}; use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
use actix_service::{IntoService, IntoServiceFactory, Service, ServiceFactory}; use actix_service::{IntoService, IntoServiceFactory, Service, ServiceFactory};
use either::Either; use either::Either;
use futures::{ready, Stream}; use futures_core::{ready, stream::Stream};
use pin_project::project;
use crate::connect::{Connect, ConnectResult}; use crate::connect::{Connect, ConnectResult};
use crate::dispatcher::Dispatcher; use crate::dispatcher::Dispatcher;
@ -336,7 +335,7 @@ where
} }
} }
#[pin_project::pin_project] #[pin_project::pin_project(project = FramedServiceImplResponseInnerProj)]
enum FramedServiceImplResponseInner<St, Io, Codec, Out, C, T> enum FramedServiceImplResponseInner<St, Io, Codec, Out, C, T>
where where
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec, Out>>, C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec, Out>>,
@ -378,7 +377,6 @@ where
<Codec as Encoder>::Error: std::fmt::Debug, <Codec as Encoder>::Error: std::fmt::Debug,
Out: Stream<Item = <Codec as Encoder>::Item> + Unpin, Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
{ {
#[project]
fn poll( fn poll(
self: Pin<&mut Self>, self: Pin<&mut Self>,
cx: &mut Context<'_>, cx: &mut Context<'_>,
@ -386,9 +384,8 @@ where
FramedServiceImplResponseInner<St, Io, Codec, Out, C, T>, FramedServiceImplResponseInner<St, Io, Codec, Out, C, T>,
Poll<Result<(), ServiceError<C::Error, Codec>>>, Poll<Result<(), ServiceError<C::Error, Codec>>>,
> { > {
#[project]
match self.project() { match self.project() {
FramedServiceImplResponseInner::Connect(fut, handler) => match fut.poll(cx) { FramedServiceImplResponseInnerProj::Connect(fut, handler) => match fut.poll(cx) {
Poll::Ready(Ok(res)) => Either::Left(FramedServiceImplResponseInner::Handler( Poll::Ready(Ok(res)) => Either::Left(FramedServiceImplResponseInner::Handler(
handler.new_service(res.state), handler.new_service(res.state),
Some(res.framed), Some(res.framed),
@ -397,7 +394,7 @@ where
Poll::Pending => Either::Right(Poll::Pending), Poll::Pending => Either::Right(Poll::Pending),
Poll::Ready(Err(e)) => Either::Right(Poll::Ready(Err(e.into()))), Poll::Ready(Err(e)) => Either::Right(Poll::Ready(Err(e.into()))),
}, },
FramedServiceImplResponseInner::Handler(fut, framed, out) => { FramedServiceImplResponseInnerProj::Handler(fut, framed, out) => {
match fut.poll(cx) { match fut.poll(cx) {
Poll::Ready(Ok(handler)) => { Poll::Ready(Ok(handler)) => {
Either::Left(FramedServiceImplResponseInner::Dispatcher( Either::Left(FramedServiceImplResponseInner::Dispatcher(
@ -408,7 +405,7 @@ where
Poll::Ready(Err(e)) => Either::Right(Poll::Ready(Err(e.into()))), Poll::Ready(Err(e)) => Either::Right(Poll::Ready(Err(e.into()))),
} }
} }
FramedServiceImplResponseInner::Dispatcher(fut) => { FramedServiceImplResponseInnerProj::Dispatcher(fut) => {
Either::Right(fut.poll(cx)) Either::Right(fut.poll(cx))
} }
} }

View File

@ -6,7 +6,7 @@ use actix_service::{fn_factory_with_config, fn_service, IntoService, Service};
use actix_testing::TestServer; use actix_testing::TestServer;
use actix_utils::mpsc; use actix_utils::mpsc;
use bytes::{Bytes, BytesMut}; use bytes::{Bytes, BytesMut};
use futures::future::ok; use futures_util::future::ok;
use actix_ioframe::{Builder, Connect, FactoryBuilder}; use actix_ioframe::{Builder, Connect, FactoryBuilder};

1
actix-macros/.gitignore vendored Normal file
View File

@ -0,0 +1 @@
/wip

9
actix-macros/CHANGES.md Normal file
View File

@ -0,0 +1,9 @@
# CHANGES
## 0.1.2 - 2020-05-18
### Changed
* Forward actix_rt::test arguments to test function [#127]
[#127]: https://github.com/actix/actix-net/pull/127

View File

@ -1,6 +1,6 @@
[package] [package]
name = "actix-macros" name = "actix-macros"
version = "0.1.1" version = "0.1.2"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"] authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix runtime macros" description = "Actix runtime macros"
repository = "https://github.com/actix/actix-net" repository = "https://github.com/actix/actix-net"
@ -8,14 +8,16 @@ documentation = "https://docs.rs/actix-macros/"
categories = ["network-programming", "asynchronous"] categories = ["network-programming", "asynchronous"]
license = "MIT/Apache-2.0" license = "MIT/Apache-2.0"
edition = "2018" edition = "2018"
workspace = ".."
[lib] [lib]
proc-macro = true proc-macro = true
[dependencies] [dependencies]
quote = "=1.0.2" quote = "1.0.3"
syn = { version = "^1", features = ["full"] } syn = { version = "^1", features = ["full"] }
[dev-dependencies] [dev-dependencies]
actix-rt = { version = "1.0.0" } actix-rt = "1.0"
futures-util = { version = "0.3", default-features = false }
trybuild = "1"

View File

@ -55,12 +55,11 @@ pub fn main(_: TokenStream, item: TokenStream) -> TokenStream {
/// ``` /// ```
#[proc_macro_attribute] #[proc_macro_attribute]
pub fn test(_: TokenStream, item: TokenStream) -> TokenStream { pub fn test(_: TokenStream, item: TokenStream) -> TokenStream {
let input = syn::parse_macro_input!(item as syn::ItemFn); let mut input = syn::parse_macro_input!(item as syn::ItemFn);
let ret = &input.sig.output;
let name = &input.sig.ident;
let body = &input.block;
let attrs = &input.attrs; let attrs = &input.attrs;
let vis = &input.vis;
let sig = &mut input.sig;
let body = &input.block;
let mut has_test_attr = false; let mut has_test_attr = false;
for attr in attrs { for attr in attrs {
@ -69,7 +68,7 @@ pub fn test(_: TokenStream, item: TokenStream) -> TokenStream {
} }
} }
if input.sig.asyncness.is_none() { if sig.asyncness.is_none() {
return syn::Error::new_spanned( return syn::Error::new_spanned(
input.sig.fn_token, input.sig.fn_token,
format!("only async fn is supported, {}", input.sig.ident), format!("only async fn is supported, {}", input.sig.ident),
@ -78,10 +77,12 @@ pub fn test(_: TokenStream, item: TokenStream) -> TokenStream {
.into(); .into();
} }
sig.asyncness = None;
let result = if has_test_attr { let result = if has_test_attr {
quote! { quote! {
#(#attrs)* #(#attrs)*
fn #name() #ret { #vis #sig {
actix_rt::System::new("test") actix_rt::System::new("test")
.block_on(async { #body }) .block_on(async { #body })
} }
@ -90,7 +91,7 @@ pub fn test(_: TokenStream, item: TokenStream) -> TokenStream {
quote! { quote! {
#[test] #[test]
#(#attrs)* #(#attrs)*
fn #name() #ret { #vis #sig {
actix_rt::System::new("test") actix_rt::System::new("test")
.block_on(async { #body }) .block_on(async { #body })
} }

View File

@ -0,0 +1,9 @@
#[test]
fn compile_macros() {
let t = trybuild::TestCases::new();
t.pass("tests/trybuild/main-01-basic.rs");
t.compile_fail("tests/trybuild/main-02-only-async.rs");
t.pass("tests/trybuild/test-01-basic.rs");
t.pass("tests/trybuild/test-02-keep-attrs.rs");
}

View File

@ -0,0 +1,4 @@
#[actix_rt::main]
async fn main() {
println!("Hello world");
}

View File

@ -0,0 +1,4 @@
#[actix_rt::main]
fn main() {
futures_util::future::ready(()).await
}

View File

@ -0,0 +1,14 @@
error: only async fn is supported
--> $DIR/main-02-only-async.rs:2:1
|
2 | fn main() {
| ^^
error[E0601]: `main` function not found in crate `$CRATE`
--> $DIR/main-02-only-async.rs:1:1
|
1 | / #[actix_rt::main]
2 | | fn main() {
3 | | futures_util::future::ready(()).await
4 | | }
| |_^ consider adding a `main` function to `$DIR/tests/trybuild/main-02-only-async.rs`

View File

@ -0,0 +1,6 @@
#[actix_rt::test]
async fn my_test() {
assert!(true);
}
fn main() {}

View File

@ -0,0 +1,7 @@
#[actix_rt::test]
#[should_panic]
async fn my_test() {
todo!()
}
fn main() {}

View File

@ -1,10 +1,27 @@
# Changes # Changes
## [TBD] - [TBD] ## [1.1.1] - 2020-04-30
- Expose `System::is_set` to check if current system is running ### Fixed
- Add `Arbiter::local_join` associated function to get be able to `await` for spawned futures * Fix memory leak due to [#94] (see [#129] for more detail)
[#129]: https://github.com/actix/actix-net/issues/129
## [1.1.0] - 2020-04-08
**This version has been yanked.**
### Added
* Expose `System::is_set` to check if current system has ben started [#99]
* Add `Arbiter::is_running` to check if event loop is running [#124]
* Add `Arbiter::local_join` associated function
to get be able to `await` for spawned futures [#94]
[#94]: https://github.com/actix/actix-net/pull/94
[#99]: https://github.com/actix/actix-net/pull/99
[#124]: https://github.com/actix/actix-net/pull/124
## [1.0.0] - 2019-12-11 ## [1.0.0] - 2019-12-11

View File

@ -1,6 +1,6 @@
[package] [package]
name = "actix-rt" name = "actix-rt"
version = "1.0.0" version = "1.1.1"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"] authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix runtime" description = "Actix runtime"
keywords = ["network", "framework", "async", "futures"] keywords = ["network", "framework", "async", "futures"]
@ -18,7 +18,8 @@ path = "src/lib.rs"
[dependencies] [dependencies]
actix-macros = "0.1.0" actix-macros = "0.1.0"
actix-threadpool = "0.3" actix-threadpool = "0.3"
futures-channel = { version = "0.3.1", default-features = false } futures-channel = { version = "0.3.4", default-features = false }
futures-util = { version = "0.3.1", default-features = false } futures-util = { version = "0.3.4", default-features = false, features = ["alloc"] }
copyless = "0.1.4" copyless = "0.1.4"
smallvec = "1"
tokio = { version = "0.2.6", default-features = false, features = ["rt-core", "rt-util", "io-driver", "tcp", "uds", "udp", "time", "signal", "stream"] } tokio = { version = "0.2.6", default-features = false, features = ["rt-core", "rt-util", "io-driver", "tcp", "uds", "udp", "time", "signal", "stream"] }

View File

@ -8,20 +8,24 @@ use std::{fmt, thread};
use futures_channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender}; use futures_channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender};
use futures_channel::oneshot::{channel, Canceled, Sender}; use futures_channel::oneshot::{channel, Canceled, Sender};
use futures_util::{future::{self, Future, FutureExt}, stream::Stream}; use futures_util::{
future::{self, Future, FutureExt},
stream::Stream,
};
use crate::runtime::Runtime; use crate::runtime::Runtime;
use crate::system::System; use crate::system::System;
use copyless::BoxHelper; use copyless::BoxHelper;
use smallvec::SmallVec;
pub use tokio::task::JoinHandle; pub use tokio::task::JoinHandle;
thread_local!( thread_local!(
static ADDR: RefCell<Option<Arbiter>> = RefCell::new(None); static ADDR: RefCell<Option<Arbiter>> = RefCell::new(None);
static RUNNING: Cell<bool> = Cell::new(false); static RUNNING: Cell<bool> = Cell::new(false);
static Q: RefCell<Vec<Pin<Box<dyn Future<Output = ()>>>>> = RefCell::new(Vec::new()); static Q: RefCell<Vec<Pin<Box<dyn Future<Output = ()>>>>> = RefCell::new(Vec::new());
static PENDING: RefCell<Vec<JoinHandle<()>>> = RefCell::new(Vec::new()); static PENDING: RefCell<SmallVec<[JoinHandle<()>; 8]>> = RefCell::new(SmallVec::new());
static STORAGE: RefCell<HashMap<TypeId, Box<dyn Any>>> = RefCell::new(HashMap::new()); static STORAGE: RefCell<HashMap<TypeId, Box<dyn Any>>> = RefCell::new(HashMap::new());
); );
@ -86,6 +90,11 @@ impl Arbiter {
}) })
} }
/// Check if current arbiter is running.
pub fn is_running() -> bool {
RUNNING.with(|cell| cell.get())
}
/// Stop arbiter from continuing it's event loop. /// Stop arbiter from continuing it's event loop.
pub fn stop(&self) { pub fn stop(&self) {
let _ = self.sender.unbounded_send(ArbiterCommand::Stop); let _ = self.sender.unbounded_send(ArbiterCommand::Stop);
@ -173,15 +182,20 @@ impl Arbiter {
RUNNING.with(move |cell| { RUNNING.with(move |cell| {
if cell.get() { if cell.get() {
// Spawn the future on running executor // Spawn the future on running executor
PENDING.with(move |cell| { let len = PENDING.with(move |cell| {
cell.borrow_mut().push(tokio::task::spawn_local(future)); let mut p = cell.borrow_mut();
}) p.push(tokio::task::spawn_local(future));
p.len()
});
if len > 7 {
// Before reaching the inline size
tokio::task::spawn_local(CleanupPending);
}
} else { } else {
// Box the future and push it to the queue, this results in double boxing // Box the future and push it to the queue, this results in double boxing
// because the executor boxes the future again, but works for now // because the executor boxes the future again, but works for now
Q.with(move |cell| { Q.with(move |cell| {
cell.borrow_mut() cell.borrow_mut().push(Pin::from(Box::alloc().init(future)))
.push(Pin::from(Box::alloc().init(future)))
}); });
} }
}); });
@ -304,12 +318,36 @@ impl Arbiter {
/// have completed. /// have completed.
pub fn local_join() -> impl Future<Output = ()> { pub fn local_join() -> impl Future<Output = ()> {
PENDING.with(move |cell| { PENDING.with(move |cell| {
let current = cell.replace(Vec::new()); let current = cell.replace(SmallVec::new());
future::join_all(current).map(|_| ()) future::join_all(current).map(|_| ())
}) })
} }
} }
/// Future used for cleaning-up already finished `JoinHandle`s
/// from the `PENDING` list so the vector doesn't grow indefinitely
struct CleanupPending;
impl Future for CleanupPending {
type Output = ();
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
PENDING.with(move |cell| {
let mut pending = cell.borrow_mut();
let mut i = 0;
while i != pending.len() {
if let Poll::Ready(_) = Pin::new(&mut pending[i]).poll(cx) {
pending.remove(i);
} else {
i += 1;
}
}
});
Poll::Ready(())
}
}
struct ArbiterController { struct ArbiterController {
stop: Option<Sender<i32>>, stop: Option<Sender<i32>>,
rx: UnboundedReceiver<ArbiterCommand>, rx: UnboundedReceiver<ArbiterCommand>,
@ -343,9 +381,15 @@ impl Future for ArbiterController {
return Poll::Ready(()); return Poll::Ready(());
} }
ArbiterCommand::Execute(fut) => { ArbiterCommand::Execute(fut) => {
PENDING.with(move |cell| { let len = PENDING.with(move |cell| {
cell.borrow_mut().push(tokio::task::spawn_local(fut)); let mut p = cell.borrow_mut();
p.push(tokio::task::spawn_local(fut));
p.len()
}); });
if len > 7 {
// Before reaching the inline size
tokio::task::spawn_local(CleanupPending);
}
} }
ArbiterCommand::ExecuteFn(f) => { ArbiterCommand::ExecuteFn(f) => {
f.call_box(); f.call_box();

View File

@ -79,7 +79,7 @@ impl System {
}) })
} }
/// Check if current system is running. /// Check if current system is set, i.e., as already been started.
pub fn is_set() -> bool { pub fn is_set() -> bool {
CURRENT.with(|cell| cell.borrow().is_some()) CURRENT.with(|cell| cell.borrow().is_some())
} }

View File

@ -1,5 +1,19 @@
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
#[test]
fn start_and_stop() {
actix_rt::System::new("start_and_stop").block_on(async move {
assert!(
actix_rt::Arbiter::is_running(),
"System doesn't seem to have started"
);
});
assert!(
!actix_rt::Arbiter::is_running(),
"System doesn't seem to have stopped"
);
}
#[test] #[test]
fn await_for_timer() { fn await_for_timer() {
let time = Duration::from_secs(2); let time = Duration::from_secs(2);

View File

@ -1,5 +1,13 @@
# Changes # Changes
## [1.0.3] - 2020-05-19
### Changed
* Replace deprecated `net2` crate with `socket2` [#140]
[#140]: https://github.com/actix/actix-net/pull/140
## [1.0.2] - 2020-02-26 ## [1.0.2] - 2020-02-26
### Fixed ### Fixed

View File

@ -1,6 +1,6 @@
[package] [package]
name = "actix-server" name = "actix-server"
version = "1.0.2" version = "1.0.3"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"] authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix server - General purpose tcp server" description = "Actix server - General purpose tcp server"
keywords = ["network", "framework", "async", "futures"] keywords = ["network", "framework", "async", "futures"]
@ -9,7 +9,7 @@ repository = "https://github.com/actix/actix-net.git"
documentation = "https://docs.rs/actix-server/" documentation = "https://docs.rs/actix-server/"
categories = ["network-programming", "asynchronous"] categories = ["network-programming", "asynchronous"]
license = "MIT/Apache-2.0" license = "MIT/Apache-2.0"
exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"] exclude = [".gitignore", ".cargo/config"]
edition = "2018" edition = "2018"
workspace = ".." workspace = ".."
@ -29,8 +29,9 @@ actix-utils = "1.0.4"
log = "0.4" log = "0.4"
num_cpus = "1.11" num_cpus = "1.11"
mio = "0.6.19" mio = "0.6.19"
net2 = "0.2" socket2 = "0.3"
futures = "0.3.1" futures-channel = { version = "0.3.4", default-features = false }
futures-util = { version = "0.3.4", default-features = false, features = ["sink"] }
slab = "0.4" slab = "0.4"
# unix domain sockets # unix domain sockets

View File

@ -6,13 +6,13 @@ use std::{io, mem, net};
use actix_rt::net::TcpStream; use actix_rt::net::TcpStream;
use actix_rt::time::{delay_until, Instant}; use actix_rt::time::{delay_until, Instant};
use actix_rt::{spawn, System}; use actix_rt::{spawn, System};
use futures::channel::mpsc::{unbounded, UnboundedReceiver}; use futures_channel::mpsc::{unbounded, UnboundedReceiver};
use futures::channel::oneshot; use futures_channel::oneshot;
use futures::future::ready; use futures_util::future::ready;
use futures::stream::FuturesUnordered; use futures_util::stream::FuturesUnordered;
use futures::{ready, Future, FutureExt, Stream, StreamExt}; use futures_util::{future::Future, ready, stream::Stream, FutureExt, StreamExt};
use log::{error, info}; use log::{error, info};
use net2::TcpBuilder; use socket2::{Domain, Protocol, Socket, Type};
use crate::accept::{AcceptLoop, AcceptNotify, Command}; use crate::accept::{AcceptLoop, AcceptNotify, Command};
use crate::config::{ConfiguredService, ServiceConfig}; use crate::config::{ConfiguredService, ServiceConfig};
@ -381,7 +381,7 @@ impl ServerBuilder {
.await; .await;
System::current().stop(); System::current().stop();
} }
.boxed(), .boxed(),
); );
} }
ready(()) ready(())
@ -487,11 +487,13 @@ pub(super) fn bind_addr<S: net::ToSocketAddrs>(
} }
fn create_tcp_listener(addr: net::SocketAddr, backlog: i32) -> io::Result<net::TcpListener> { fn create_tcp_listener(addr: net::SocketAddr, backlog: i32) -> io::Result<net::TcpListener> {
let builder = match addr { let domain = match addr {
net::SocketAddr::V4(_) => TcpBuilder::new_v4()?, net::SocketAddr::V4(_) => Domain::ipv4(),
net::SocketAddr::V6(_) => TcpBuilder::new_v6()?, net::SocketAddr::V6(_) => Domain::ipv6(),
}; };
builder.reuse_address(true)?; let socket = Socket::new(domain, Type::stream(), Some(Protocol::tcp()))?;
builder.bind(addr)?; socket.set_reuse_address(true)?;
Ok(builder.listen(backlog)?) socket.bind(&addr.into())?;
socket.listen(backlog)?;
Ok(socket.into_tcp_listener())
} }

View File

@ -4,7 +4,7 @@ use std::{fmt, io, net};
use actix_rt::net::TcpStream; use actix_rt::net::TcpStream;
use actix_service as actix; use actix_service as actix;
use actix_utils::counter::CounterGuard; use actix_utils::counter::CounterGuard;
use futures::future::{ok, Future, FutureExt, LocalBoxFuture}; use futures_util::future::{ok, Future, FutureExt, LocalBoxFuture};
use log::error; use log::error;
use super::builder::bind_addr; use super::builder::bind_addr;
@ -218,7 +218,7 @@ impl ServiceRuntime {
// let name = name.to_owned(); // let name = name.to_owned();
if let Some(token) = self.names.get(name) { if let Some(token) = self.names.get(name) {
self.services.insert( self.services.insert(
token.clone(), *token,
Box::new(ServiceFactory { Box::new(ServiceFactory {
inner: service.into_factory(), inner: service.into_factory(),
}), }),

View File

@ -3,9 +3,9 @@ use std::io;
use std::pin::Pin; use std::pin::Pin;
use std::task::{Context, Poll}; use std::task::{Context, Poll};
use futures::channel::mpsc::UnboundedSender; use futures_channel::mpsc::UnboundedSender;
use futures::channel::oneshot; use futures_channel::oneshot;
use futures::FutureExt; use futures_util::FutureExt;
use crate::builder::ServerBuilder; use crate::builder::ServerBuilder;
use crate::signals::Signal; use crate::signals::Signal;

View File

@ -6,8 +6,8 @@ use std::time::Duration;
use actix_rt::spawn; use actix_rt::spawn;
use actix_service::{self as actix, Service, ServiceFactory as ActixServiceFactory}; use actix_service::{self as actix, Service, ServiceFactory as ActixServiceFactory};
use actix_utils::counter::CounterGuard; use actix_utils::counter::CounterGuard;
use futures::future::{err, ok, LocalBoxFuture, Ready}; use futures_util::future::{err, ok, LocalBoxFuture, Ready};
use futures::{FutureExt, TryFutureExt}; use futures_util::{FutureExt, TryFutureExt};
use log::error; use log::error;
use super::Token; use super::Token;

View File

@ -3,7 +3,7 @@ use std::io;
use std::pin::Pin; use std::pin::Pin;
use std::task::{Context, Poll}; use std::task::{Context, Poll};
use futures::future::lazy; use futures_util::future::lazy;
use crate::server::Server; use crate::server::Server;

View File

@ -7,10 +7,10 @@ use std::time;
use actix_rt::time::{delay_until, Delay, Instant}; use actix_rt::time::{delay_until, Delay, Instant};
use actix_rt::{spawn, Arbiter}; use actix_rt::{spawn, Arbiter};
use actix_utils::counter::Counter; use actix_utils::counter::Counter;
use futures::channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender}; use futures_channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender};
use futures::channel::oneshot; use futures_channel::oneshot;
use futures::future::{join_all, LocalBoxFuture, MapOk}; use futures_util::future::{join_all, LocalBoxFuture, MapOk};
use futures::{Future, FutureExt, Stream, TryFutureExt}; use futures_util::{future::Future, stream::Stream, FutureExt, TryFutureExt};
use log::{error, info, trace}; use log::{error, info, trace};
use crate::accept::AcceptNotify; use crate::accept::AcceptNotify;

View File

@ -4,15 +4,15 @@ use std::{net, thread, time};
use actix_server::Server; use actix_server::Server;
use actix_service::fn_service; use actix_service::fn_service;
use futures::future::{lazy, ok}; use futures_util::future::{lazy, ok};
use net2::TcpBuilder; use socket2::{Domain, Protocol, Socket, Type};
fn unused_addr() -> net::SocketAddr { fn unused_addr() -> net::SocketAddr {
let addr: net::SocketAddr = "127.0.0.1:0".parse().unwrap(); let addr: net::SocketAddr = "127.0.0.1:0".parse().unwrap();
let socket = TcpBuilder::new_v4().unwrap(); let socket = Socket::new(Domain::ipv4(), Type::stream(), Some(Protocol::tcp())).unwrap();
socket.bind(&addr).unwrap(); socket.bind(&addr.into()).unwrap();
socket.reuse_address(true).unwrap(); socket.set_reuse_address(true).unwrap();
let tcp = socket.to_tcp_listener().unwrap(); let tcp = socket.into_tcp_listener();
tcp.local_addr().unwrap() tcp.local_addr().unwrap()
} }
@ -71,7 +71,7 @@ fn test_start() {
use actix_codec::{BytesCodec, Framed}; use actix_codec::{BytesCodec, Framed};
use actix_rt::net::TcpStream; use actix_rt::net::TcpStream;
use bytes::Bytes; use bytes::Bytes;
use futures::SinkExt; use futures_util::sink::SinkExt;
use std::io::Read; use std::io::Read;
let addr = unused_addr(); let addr = unused_addr();
@ -83,12 +83,10 @@ fn test_start() {
.backlog(100) .backlog(100)
.disable_signals() .disable_signals()
.bind("test", addr, move || { .bind("test", addr, move || {
fn_service(|io: TcpStream| { fn_service(|io: TcpStream| async move {
async move { let mut f = Framed::new(io, BytesCodec);
let mut f = Framed::new(io, BytesCodec); f.send(Bytes::from_static(b"test")).await.unwrap();
f.send(Bytes::from_static(b"test")).await.unwrap(); Ok::<_, ()>(())
Ok::<_, ()>(())
}
}) })
}) })
.unwrap() .unwrap()

View File

@ -17,7 +17,7 @@ path = "src/lib.rs"
[dependencies] [dependencies]
futures-util = "0.3.1" futures-util = "0.3.1"
pin-project = "0.4.6" pin-project = "0.4.17"
[dev-dependencies] [dev-dependencies]
actix-rt = "1.0.0" actix-rt = "1.0.0"

View File

@ -81,7 +81,7 @@ where
state: State<A, B>, state: State<A, B>,
} }
#[pin_project::pin_project] #[pin_project::pin_project(project = StateProj)]
enum State<A, B> enum State<A, B>
where where
A: Service, A: Service,
@ -99,13 +99,11 @@ where
{ {
type Output = Result<B::Response, A::Error>; type Output = Result<B::Response, A::Error>;
#[pin_project::project]
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.as_mut().project(); let mut this = self.as_mut().project();
#[project]
match this.state.as_mut().project() { match this.state.as_mut().project() {
State::A(fut, b) => match fut.poll(cx)? { StateProj::A(fut, b) => match fut.poll(cx)? {
Poll::Ready(res) => { Poll::Ready(res) => {
let b = b.take().unwrap(); let b = b.take().unwrap();
this.state.set(State::Empty); // drop fut A this.state.set(State::Empty); // drop fut A
@ -115,11 +113,11 @@ where
} }
Poll::Pending => Poll::Pending, Poll::Pending => Poll::Pending,
}, },
State::B(fut) => fut.poll(cx).map(|r| { StateProj::B(fut) => fut.poll(cx).map(|r| {
this.state.set(State::Empty); this.state.set(State::Empty);
r r
}), }),
State::Empty => panic!("future must not be polled after it returned `Poll::Ready`"), StateProj::Empty => panic!("future must not be polled after it returned `Poll::Ready`"),
} }
} }
} }
@ -179,7 +177,7 @@ where
state: StateRC<A, B>, state: StateRC<A, B>,
} }
#[pin_project::pin_project] #[pin_project::pin_project(project = StateRCProj)]
enum StateRC<A, B> enum StateRC<A, B>
where where
A: Service, A: Service,
@ -197,13 +195,11 @@ where
{ {
type Output = Result<B::Response, A::Error>; type Output = Result<B::Response, A::Error>;
#[pin_project::project]
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.as_mut().project(); let mut this = self.as_mut().project();
#[project]
match this.state.as_mut().project() { match this.state.as_mut().project() {
StateRC::A(fut, b) => match fut.poll(cx)? { StateRCProj::A(fut, b) => match fut.poll(cx)? {
Poll::Ready(res) => { Poll::Ready(res) => {
let b = b.take().unwrap(); let b = b.take().unwrap();
this.state.set(StateRC::Empty); // drop fut A this.state.set(StateRC::Empty); // drop fut A
@ -213,11 +209,11 @@ where
} }
Poll::Pending => Poll::Pending, Poll::Pending => Poll::Pending,
}, },
StateRC::B(fut) => fut.poll(cx).map(|r| { StateRCProj::B(fut) => fut.poll(cx).map(|r| {
this.state.set(StateRC::Empty); this.state.set(StateRC::Empty);
r r
}), }),
StateRC::Empty => panic!("future must not be polled after it returned `Poll::Ready`"), StateRCProj::Empty => panic!("future must not be polled after it returned `Poll::Ready`"),
} }
} }
} }

View File

@ -67,7 +67,7 @@ where
state: State<A, B>, state: State<A, B>,
} }
#[pin_project::pin_project] #[pin_project::pin_project(project = StateProj)]
enum State<A, B> enum State<A, B>
where where
A: Service, A: Service,
@ -85,13 +85,11 @@ where
{ {
type Output = Result<B::Response, A::Error>; type Output = Result<B::Response, A::Error>;
#[pin_project::project]
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.as_mut().project(); let mut this = self.as_mut().project();
#[project]
match this.state.as_mut().project() { match this.state.as_mut().project() {
State::A(fut, b) => match fut.poll(cx)? { StateProj::A(fut, b) => match fut.poll(cx)? {
Poll::Ready(res) => { Poll::Ready(res) => {
let mut b = b.take().unwrap(); let mut b = b.take().unwrap();
this.state.set(State::Empty); // drop fut A this.state.set(State::Empty); // drop fut A
@ -101,11 +99,11 @@ where
} }
Poll::Pending => Poll::Pending, Poll::Pending => Poll::Pending,
}, },
State::B(fut) => fut.poll(cx).map(|r| { StateProj::B(fut) => fut.poll(cx).map(|r| {
this.state.set(State::Empty); this.state.set(State::Empty);
r r
}), }),
State::Empty => panic!("future must not be polled after it returned `Poll::Ready`"), StateProj::Empty => panic!("future must not be polled after it returned `Poll::Ready`"),
} }
} }
} }

View File

@ -98,7 +98,7 @@ where
state: State<A, B, F, Fut, Res, Err>, state: State<A, B, F, Fut, Res, Err>,
} }
#[pin_project::pin_project] #[pin_project::pin_project(project = StateProj)]
enum State<A, B, F, Fut, Res, Err> enum State<A, B, F, Fut, Res, Err>
where where
A: Service, A: Service,
@ -123,13 +123,11 @@ where
{ {
type Output = Result<Res, Err>; type Output = Result<Res, Err>;
#[pin_project::project]
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.as_mut().project(); let mut this = self.as_mut().project();
#[project]
match this.state.as_mut().project() { match this.state.as_mut().project() {
State::A(fut, b) => match fut.poll(cx)? { StateProj::A(fut, b) => match fut.poll(cx)? {
Poll::Ready(res) => { Poll::Ready(res) => {
let mut b = b.take().unwrap(); let mut b = b.take().unwrap();
this.state.set(State::Empty); this.state.set(State::Empty);
@ -140,11 +138,11 @@ where
} }
Poll::Pending => Poll::Pending, Poll::Pending => Poll::Pending,
}, },
State::B(fut) => fut.poll(cx).map(|r| { StateProj::B(fut) => fut.poll(cx).map(|r| {
this.state.set(State::Empty); this.state.set(State::Empty);
r r
}), }),
State::Empty => panic!("future must not be polled after it returned `Poll::Ready`"), StateProj::Empty => panic!("future must not be polled after it returned `Poll::Ready`"),
} }
} }
} }

View File

@ -177,7 +177,7 @@ where
state: State<T, R, S>, state: State<T, R, S>,
} }
#[pin_project::pin_project] #[pin_project::pin_project(project = StateProj)]
enum State<T, R, S> enum State<T, R, S>
where where
T: ServiceFactory<Config = ()>, T: ServiceFactory<Config = ()>,
@ -200,20 +200,18 @@ where
{ {
type Output = Result<S, T::InitError>; type Output = Result<S, T::InitError>;
#[pin_project::project]
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.as_mut().project(); let mut this = self.as_mut().project();
#[project]
match this.state.as_mut().project() { match this.state.as_mut().project() {
State::A(fut) => match fut.poll(cx)? { StateProj::A(fut) => match fut.poll(cx)? {
Poll::Pending => Poll::Pending, Poll::Pending => Poll::Pending,
Poll::Ready(srv) => { Poll::Ready(srv) => {
this.state.set(State::B(srv)); this.state.set(State::B(srv));
self.poll(cx) self.poll(cx)
} }
}, },
State::B(srv) => match srv.poll_ready(cx)? { StateProj::B(srv) => match srv.poll_ready(cx)? {
Poll::Ready(_) => { Poll::Ready(_) => {
let fut = (this.store.get_mut().1)(this.cfg.take().unwrap(), srv); let fut = (this.store.get_mut().1)(this.cfg.take().unwrap(), srv);
this.state.set(State::C(fut)); this.state.set(State::C(fut));
@ -221,7 +219,7 @@ where
} }
Poll::Pending => Poll::Pending, Poll::Pending => Poll::Pending,
}, },
State::C(fut) => fut.poll(cx), StateProj::C(fut) => fut.poll(cx),
} }
} }
} }

View File

@ -66,7 +66,7 @@ where
state: State<A, B>, state: State<A, B>,
} }
#[pin_project::pin_project] #[pin_project::pin_project(project = StateProj)]
enum State<A, B> enum State<A, B>
where where
A: Service, A: Service,
@ -84,13 +84,11 @@ where
{ {
type Output = Result<B::Response, B::Error>; type Output = Result<B::Response, B::Error>;
#[pin_project::project]
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.as_mut().project(); let mut this = self.as_mut().project();
#[project]
match this.state.as_mut().project() { match this.state.as_mut().project() {
State::A(fut, b) => match fut.poll(cx) { StateProj::A(fut, b) => match fut.poll(cx) {
Poll::Ready(res) => { Poll::Ready(res) => {
let mut b = b.take().unwrap(); let mut b = b.take().unwrap();
this.state.set(State::Empty); // drop fut A this.state.set(State::Empty); // drop fut A
@ -100,11 +98,11 @@ where
} }
Poll::Pending => Poll::Pending, Poll::Pending => Poll::Pending,
}, },
State::B(fut) => fut.poll(cx).map(|r| { StateProj::B(fut) => fut.poll(cx).map(|r| {
this.state.set(State::Empty); this.state.set(State::Empty);
r r
}), }),
State::Empty => panic!("future must not be polled after it returned `Poll::Ready`"), StateProj::Empty => panic!("future must not be polled after it returned `Poll::Ready`"),
} }
} }
} }

View File

@ -211,7 +211,7 @@ where
state: ApplyTransformFutureState<T, S>, state: ApplyTransformFutureState<T, S>,
} }
#[pin_project::pin_project] #[pin_project::pin_project(project = ApplyTransformFutureStateProj)]
pub enum ApplyTransformFutureState<T, S> pub enum ApplyTransformFutureState<T, S>
where where
S: ServiceFactory, S: ServiceFactory,
@ -228,13 +228,11 @@ where
{ {
type Output = Result<T::Transform, T::InitError>; type Output = Result<T::Transform, T::InitError>;
#[pin_project::project]
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.as_mut().project(); let mut this = self.as_mut().project();
#[project]
match this.state.as_mut().project() { match this.state.as_mut().project() {
ApplyTransformFutureState::A(fut) => match fut.poll(cx)? { ApplyTransformFutureStateProj::A(fut) => match fut.poll(cx)? {
Poll::Ready(srv) => { Poll::Ready(srv) => {
let fut = this.store.0.new_transform(srv); let fut = this.store.0.new_transform(srv);
this.state.set(ApplyTransformFutureState::B(fut)); this.state.set(ApplyTransformFutureState::B(fut));
@ -242,7 +240,7 @@ where
} }
Poll::Pending => Poll::Pending, Poll::Pending => Poll::Pending,
}, },
ApplyTransformFutureState::B(fut) => fut.poll(cx), ApplyTransformFutureStateProj::B(fut) => fut.poll(cx),
} }
} }
} }

View File

@ -1,5 +1,11 @@
# Changes # Changes
## [1.0.1] - 2020-05-19
* Replace deprecated `net2` crate with `socket2`
* Remove unused `futures` dependency
## [1.0.0] - 2019-12-11 ## [1.0.0] - 2019-12-11
* Update actix-server to 1.0.0 * Update actix-server to 1.0.0

View File

@ -1,6 +1,6 @@
[package] [package]
name = "actix-testing" name = "actix-testing"
version = "1.0.0" version = "1.0.1"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"] authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix testing utils" description = "Actix testing utils"
keywords = ["network", "framework", "async", "futures"] keywords = ["network", "framework", "async", "futures"]
@ -11,6 +11,7 @@ categories = ["network-programming", "asynchronous"]
license = "MIT/Apache-2.0" license = "MIT/Apache-2.0"
edition = "2018" edition = "2018"
workspace = ".." workspace = ".."
readme = "README.md"
[lib] [lib]
name = "actix_testing" name = "actix_testing"
@ -23,5 +24,4 @@ actix-server = "1.0.0"
actix-service = "1.0.0" actix-service = "1.0.0"
log = "0.4" log = "0.4"
net2 = "0.2" socket2 = "0.3"
futures = "0.3.1"

View File

@ -7,7 +7,7 @@ use std::{net, thread};
use actix_rt::{net::TcpStream, System}; use actix_rt::{net::TcpStream, System};
use actix_server::{Server, ServerBuilder, ServiceFactory}; use actix_server::{Server, ServerBuilder, ServiceFactory};
use net2::TcpBuilder; use socket2::{Domain, Protocol, Socket, Type};
#[cfg(not(test))] // Work around for rust-lang/rust#62127 #[cfg(not(test))] // Work around for rust-lang/rust#62127
pub use actix_macros::test; pub use actix_macros::test;
@ -110,10 +110,11 @@ impl TestServer {
/// Get firat available unused local address /// Get firat available unused local address
pub fn unused_addr() -> net::SocketAddr { pub fn unused_addr() -> net::SocketAddr {
let addr: net::SocketAddr = "127.0.0.1:0".parse().unwrap(); let addr: net::SocketAddr = "127.0.0.1:0".parse().unwrap();
let socket = TcpBuilder::new_v4().unwrap(); let socket =
socket.bind(&addr).unwrap(); Socket::new(Domain::ipv4(), Type::stream(), Some(Protocol::tcp())).unwrap();
socket.reuse_address(true).unwrap(); socket.bind(&addr.into()).unwrap();
let tcp = socket.to_tcp_listener().unwrap(); socket.set_reuse_address(true).unwrap();
let tcp = socket.into_tcp_listener();
tcp.local_addr().unwrap() tcp.local_addr().unwrap()
} }
} }

View File

@ -1,5 +1,13 @@
# Changes # Changes
## [0.3.2] - 2020-05-20
## Added
* Implement `std::error::Error` for `BlockingError` [#120]
[#120]: https://github.com/actix/actix-net/pull/120
## [0.3.1] - 2019-12-12 ## [0.3.1] - 2019-12-12
### Changed ### Changed

View File

@ -1,6 +1,6 @@
[package] [package]
name = "actix-threadpool" name = "actix-threadpool"
version = "0.3.1" version = "0.3.2"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"] authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix thread pool for sync code" description = "Actix thread pool for sync code"
keywords = ["actix", "network", "framework", "async", "futures"] keywords = ["actix", "network", "framework", "async", "futures"]

View File

@ -48,6 +48,8 @@ pub enum BlockingError<E: fmt::Debug> {
Canceled, Canceled,
} }
impl<E: fmt::Debug> std::error::Error for BlockingError<E> {}
/// Execute blocking function on a thread pool, returns future that resolves /// Execute blocking function on a thread pool, returns future that resolves
/// to result of the function execution. /// to result of the function execution.
pub fn run<F, I, E>(f: F) -> CpuFuture<I, E> pub fn run<F, I, E>(f: F) -> CpuFuture<I, E>

View File

@ -38,7 +38,7 @@ actix-utils = "1.0.0"
actix-rt = "1.0.0" actix-rt = "1.0.0"
derive_more = "0.99.2" derive_more = "0.99.2"
either = "1.5.2" either = "1.5.2"
futures = "0.3.1" futures-util = { version = "0.3.4", default-features = false }
log = "0.4" log = "0.4"
# openssl # openssl

View File

@ -4,7 +4,7 @@ use std::task::{Context, Poll};
use actix_codec::{AsyncRead, AsyncWrite}; use actix_codec::{AsyncRead, AsyncWrite};
use actix_service::{Service, ServiceFactory}; use actix_service::{Service, ServiceFactory};
use actix_utils::counter::Counter; use actix_utils::counter::Counter;
use futures::future::{self, FutureExt, LocalBoxFuture, TryFutureExt}; use futures_util::future::{self, FutureExt, LocalBoxFuture, TryFutureExt};
pub use native_tls::Error; pub use native_tls::Error;
pub use tokio_tls::{TlsAcceptor, TlsStream}; pub use tokio_tls::{TlsAcceptor, TlsStream};

View File

@ -9,7 +9,7 @@ pub use tokio_openssl::{HandshakeError, SslStream};
use actix_codec::{AsyncRead, AsyncWrite}; use actix_codec::{AsyncRead, AsyncWrite};
use actix_service::{Service, ServiceFactory}; use actix_service::{Service, ServiceFactory};
use actix_utils::counter::{Counter, CounterGuard}; use actix_utils::counter::{Counter, CounterGuard};
use futures::future::{ok, FutureExt, LocalBoxFuture, Ready}; use futures_util::future::{ok, FutureExt, LocalBoxFuture, Ready};
use crate::MAX_CONN_COUNTER; use crate::MAX_CONN_COUNTER;
@ -105,7 +105,7 @@ impl<T: AsyncRead + AsyncWrite + Unpin> Future for AcceptorServiceResponse<T> {
type Output = Result<SslStream<T>, HandshakeError<T>>; type Output = Result<SslStream<T>, HandshakeError<T>>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let io = futures::ready!(Pin::new(&mut self.fut).poll(cx))?; let io = futures_util::ready!(Pin::new(&mut self.fut).poll(cx))?;
Poll::Ready(Ok(io)) Poll::Ready(Ok(io))
} }
} }

View File

@ -8,7 +8,7 @@ use std::task::{Context, Poll};
use actix_codec::{AsyncRead, AsyncWrite}; use actix_codec::{AsyncRead, AsyncWrite};
use actix_service::{Service, ServiceFactory}; use actix_service::{Service, ServiceFactory};
use actix_utils::counter::{Counter, CounterGuard}; use actix_utils::counter::{Counter, CounterGuard};
use futures::future::{ok, Ready}; use futures_util::future::{ok, Ready};
use tokio_rustls::{Accept, TlsAcceptor}; use tokio_rustls::{Accept, TlsAcceptor};
pub use rust_tls::{ServerConfig, Session}; pub use rust_tls::{ServerConfig, Session};
@ -108,7 +108,7 @@ impl<T: AsyncRead + AsyncWrite + Unpin> Future for AcceptorServiceFut<T> {
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.get_mut(); let this = self.get_mut();
let res = futures::ready!(Pin::new(&mut this.fut).poll(cx)); let res = futures_util::ready!(Pin::new(&mut this.fut).poll(cx));
match res { match res {
Ok(io) => Poll::Ready(Ok(io)), Ok(io) => Poll::Ready(Ok(io)),
Err(e) => Poll::Ready(Err(e)), Err(e) => Poll::Ready(Err(e)),

View File

@ -17,7 +17,7 @@ path = "src/lib.rs"
[dependencies] [dependencies]
actix-service = "1.0.4" actix-service = "1.0.4"
futures-util = "0.3.1" futures-util = { version = "0.3.4", default-features = false }
tracing = "0.1" tracing = "0.1"
tracing-futures = "0.2" tracing-futures = "0.2"

View File

@ -22,7 +22,9 @@ actix-codec = "0.2.0"
bitflags = "1.2" bitflags = "1.2"
bytes = "0.5.3" bytes = "0.5.3"
either = "1.5.3" either = "1.5.3"
futures = "0.3.1" futures-channel = { version = "0.3.4", default-features = false }
pin-project = "0.4.6" futures-sink = { version = "0.3.4", default-features = false }
futures-util = { version = "0.3.4", default-features = false }
pin-project = "0.4.17"
log = "0.4" log = "0.4"
slab = "0.4" slab = "0.4"

View File

@ -96,7 +96,7 @@ impl Drop for Waiter {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
use futures::future::lazy; use futures_util::future::lazy;
#[actix_rt::test] #[actix_rt::test]
async fn test_condition() { async fn test_condition() {

View File

@ -3,7 +3,7 @@ use std::pin::Pin;
use std::task::{Context, Poll}; use std::task::{Context, Poll};
use actix_service::{Service, ServiceFactory}; use actix_service::{Service, ServiceFactory};
use futures::{future, ready, Future}; use futures_util::{future, ready, future::Future};
/// Combine two different service types into a single type. /// Combine two different service types into a single type.
/// ///

View File

@ -6,7 +6,7 @@ use std::{fmt, mem};
use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed}; use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
use actix_service::{IntoService, Service}; use actix_service::{IntoService, Service};
use futures::{Future, FutureExt, Stream}; use futures_util::{future::Future, FutureExt, stream::Stream};
use log::debug; use log::debug;
use crate::mpsc; use crate::mpsc;

View File

@ -4,7 +4,7 @@ use std::pin::Pin;
use std::task::{Context, Poll}; use std::task::{Context, Poll};
use actix_service::{IntoService, Service, Transform}; use actix_service::{IntoService, Service, Transform};
use futures::future::{ok, Ready}; use futures_util::future::{ok, Ready};
use super::counter::{Counter, CounterGuard}; use super::counter::{Counter, CounterGuard};
@ -116,7 +116,7 @@ mod tests {
use super::*; use super::*;
use actix_service::{apply, fn_factory, Service, ServiceFactory}; use actix_service::{apply, fn_factory, Service, ServiceFactory};
use futures::future::{lazy, ok, FutureExt, LocalBoxFuture}; use futures_util::future::{lazy, ok, FutureExt, LocalBoxFuture};
struct SleepService(Duration); struct SleepService(Duration);

View File

@ -7,7 +7,7 @@ use std::time::Duration;
use actix_rt::time::{delay_until, Delay, Instant}; use actix_rt::time::{delay_until, Delay, Instant};
use actix_service::{Service, ServiceFactory}; use actix_service::{Service, ServiceFactory};
use futures::future::{ok, Ready}; use futures_util::future::{ok, Ready};
use super::time::{LowResTime, LowResTimeService}; use super::time::{LowResTime, LowResTimeService};

View File

@ -6,7 +6,8 @@ use std::fmt;
use std::pin::Pin; use std::pin::Pin;
use std::task::{Context, Poll}; use std::task::{Context, Poll};
use futures::{Sink, Stream}; use futures_sink::Sink;
use futures_util::stream::Stream;
use crate::cell::Cell; use crate::cell::Cell;
use crate::task::LocalWaker; use crate::task::LocalWaker;
@ -180,8 +181,8 @@ impl<T> SendError<T> {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
use futures::future::lazy; use futures_util::future::lazy;
use futures::{Stream, StreamExt}; use futures_util::{stream::Stream, StreamExt};
#[actix_rt::test] #[actix_rt::test]
async fn test_mpsc() { async fn test_mpsc() {

View File

@ -3,7 +3,7 @@ use std::future::Future;
use std::pin::Pin; use std::pin::Pin;
use std::task::{Context, Poll}; use std::task::{Context, Poll};
pub use futures::channel::oneshot::Canceled; pub use futures_channel::oneshot::Canceled;
use slab::Slab; use slab::Slab;
use crate::cell::Cell; use crate::cell::Cell;
@ -253,7 +253,7 @@ impl<T> Future for PReceiver<T> {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
use futures::future::lazy; use futures_util::future::lazy;
#[actix_rt::test] #[actix_rt::test]
async fn test_oneshot() { async fn test_oneshot() {

View File

@ -8,7 +8,7 @@ use std::rc::Rc;
use std::task::{Context, Poll}; use std::task::{Context, Poll};
use actix_service::{IntoService, Service, Transform}; use actix_service::{IntoService, Service, Transform};
use futures::future::{ok, Ready}; use futures_util::future::{ok, Ready};
use crate::oneshot; use crate::oneshot;
use crate::task::LocalWaker; use crate::task::LocalWaker;
@ -210,8 +210,8 @@ mod tests {
use super::*; use super::*;
use actix_service::Service; use actix_service::Service;
use futures::channel::oneshot; use futures_channel::oneshot;
use futures::future::{lazy, poll_fn, FutureExt, LocalBoxFuture}; use futures_util::future::{lazy, poll_fn, FutureExt, LocalBoxFuture};
struct Srv; struct Srv;

View File

@ -3,7 +3,7 @@ use std::pin::Pin;
use std::task::{Context, Poll}; use std::task::{Context, Poll};
use actix_service::{IntoService, Service}; use actix_service::{IntoService, Service};
use futures::{FutureExt, Stream}; use futures_util::{FutureExt, stream::Stream};
use crate::mpsc; use crate::mpsc;

View File

@ -4,7 +4,7 @@ use std::time::{self, Duration, Instant};
use actix_rt::time::delay_for; use actix_rt::time::delay_for;
use actix_service::{Service, ServiceFactory}; use actix_service::{Service, ServiceFactory};
use futures::future::{ok, ready, FutureExt, Ready}; use futures_util::future::{ok, ready, FutureExt, Ready};
use super::cell::Cell; use super::cell::Cell;

View File

@ -10,7 +10,7 @@ use std::{fmt, time};
use actix_rt::time::{delay_for, Delay}; use actix_rt::time::{delay_for, Delay};
use actix_service::{IntoService, Service, Transform}; use actix_service::{IntoService, Service, Transform};
use futures::future::{ok, Ready}; use futures_util::future::{ok, Ready};
/// Applies a timeout to requests. /// Applies a timeout to requests.
#[derive(Debug)] #[derive(Debug)]
@ -183,7 +183,7 @@ mod tests {
use super::*; use super::*;
use actix_service::{apply, fn_factory, Service, ServiceFactory}; use actix_service::{apply, fn_factory, Service, ServiceFactory};
use futures::future::{ok, FutureExt, LocalBoxFuture}; use futures_util::future::{ok, FutureExt, LocalBoxFuture};
struct SleepService(Duration); struct SleepService(Duration);

14
codecov.yml Normal file
View File

@ -0,0 +1,14 @@
coverage:
status:
project:
default:
threshold: 10% # make CI green
patch:
default:
threshold: 10% # make CI green
ignore: # ignore codecoverage on following paths
- "examples"
- ".github"
- "**/*.md"
- "**/*.toml"

View File

@ -11,7 +11,7 @@ use actix_codec::{AsyncRead, AsyncWrite};
use actix_rt::System; use actix_rt::System;
use actix_server::{Io, Server}; use actix_server::{Io, Server};
use actix_service::{service_fn, NewService}; use actix_service::{service_fn, NewService};
use futures::{future, Future}; use futures_util::{future, Future};
use openssl::ssl::{SslAcceptor, SslFiletype, SslMethod}; use openssl::ssl::{SslAcceptor, SslFiletype, SslMethod};
use tokio_openssl::SslAcceptorExt; use tokio_openssl::SslAcceptorExt;

View File

@ -7,7 +7,7 @@ use std::sync::{
use actix_rt::System; use actix_rt::System;
use actix_server::{ssl, Server}; use actix_server::{ssl, Server};
use actix_service::NewService; use actix_service::NewService;
use futures::future; use futures_util::future;
use openssl::ssl::{SslAcceptor, SslFiletype, SslMethod}; use openssl::ssl::{SslAcceptor, SslFiletype, SslMethod};
#[derive(Debug)] #[derive(Debug)]

View File

@ -31,7 +31,7 @@ fn set_bit(array: &mut [u8], ch: u8) {
} }
thread_local! { thread_local! {
static DEFAULT_QUOTER: Quoter = { Quoter::new(b"@:", b"/+") }; static DEFAULT_QUOTER: Quoter = Quoter::new(b"@:", b"/+");
} }
#[derive(Default, Clone, Debug)] #[derive(Default, Clone, Debug)]

View File

@ -1,5 +1,9 @@
# Changes # Changes
## [0.1.5] - 2020-03-30
* Serde support
## [0.1.4] - 2020-01-14 ## [0.1.4] - 2020-01-14
* Fix `AsRef<str>` impl * Fix `AsRef<str>` impl

View File

@ -1,6 +1,6 @@
[package] [package]
name = "bytestring" name = "bytestring"
version = "0.1.4" version = "0.1.5"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"] authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "A UTF-8 encoded string with Bytes as a storage" description = "A UTF-8 encoded string with Bytes as a storage"
keywords = ["actix"] keywords = ["actix"]
@ -16,3 +16,7 @@ path = "src/lib.rs"
[dependencies] [dependencies]
bytes = "0.5.3" bytes = "0.5.3"
serde = { version = "1.0", optional = true }
[dev-dependencies]
serde_json = "1.0"

View File

@ -1,10 +1,11 @@
//! A utl-8 encoded read-only string with Bytes as a storage. //! A UTF-8 encoded read-only string using Bytes as storage.
use std::convert::TryFrom; use std::convert::TryFrom;
use std::{borrow, fmt, hash, ops, str}; use std::{borrow, fmt, hash, ops, str};
use bytes::Bytes; use bytes::Bytes;
/// A utf-8 encoded string with [`Bytes`] as a storage. /// A UTF-8 encoded string with [`Bytes`] as a storage.
/// ///
/// [`Bytes`]: https://docs.rs/bytes/0.5.3/bytes/struct.Bytes.html /// [`Bytes`]: https://docs.rs/bytes/0.5.3/bytes/struct.Bytes.html
#[derive(Clone, Eq, Ord, PartialOrd, Default)] #[derive(Clone, Eq, Ord, PartialOrd, Default)]
@ -159,6 +160,34 @@ impl fmt::Display for ByteString {
} }
} }
#[cfg(feature = "serde")]
mod serde {
use serde::de::{Deserialize, Deserializer};
use serde::ser::{Serialize, Serializer};
use super::ByteString;
impl Serialize for ByteString {
#[inline]
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
serializer.serialize_str(self.as_ref())
}
}
impl<'de> Deserialize<'de> for ByteString {
#[inline]
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
String::deserialize(deserializer).map(ByteString::from)
}
}
}
#[cfg(test)] #[cfg(test)]
mod test { mod test {
use super::*; use super::*;
@ -222,4 +251,18 @@ mod test {
fn test_try_from_bytesmut() { fn test_try_from_bytesmut() {
let _ = ByteString::try_from(bytes::BytesMut::from(&b"nice bytes"[..])).unwrap(); let _ = ByteString::try_from(bytes::BytesMut::from(&b"nice bytes"[..])).unwrap();
} }
#[cfg(feature = "serde")]
#[test]
fn test_serialize() {
let s: ByteString = serde_json::from_str(r#""nice bytes""#).unwrap();
assert_eq!(s, "nice bytes");
}
#[cfg(feature = "serde")]
#[test]
fn test_deserialize() {
let s = serde_json::to_string(&ByteString::from_static("nice bytes")).unwrap();
assert_eq!(s, r#""nice bytes""#);
}
} }