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

merge master into branch mio-0.7.3

This commit is contained in:
fakeshadow 2020-12-28 05:51:31 +08:00
parent e3d43c3f36 ba44ea7d0b
commit a5c75e0feb
50 changed files with 790 additions and 1506 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
.github/workflows/bench.yml vendored
View File

@ -1,29 +0,0 @@
name: Benchmark (Linux)
on:
pull_request:
types: [opened, synchronize, reopened]
push:
branches:
- master
- '1.0'
jobs:
check_benchmark:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Install Rust
uses: actions-rs/toolchain@v1
with:
toolchain: nightly
profile: minimal
override: true
- name: Check benchmark
uses: actions-rs/cargo@v1
with:
command: bench
args: --package=actix-service

4
Cargo.toml
View File

@ -31,4 +31,6 @@ actix-router = { path = "router" }
bytestring = { path = "string" }
#FIXME: remove override
http = { git = "https://github.com/fakeshadow/http.git" }
#http = { git = "https://github.com/fakeshadow/http.git" }
trust-dns-proto = { git = "https://github.com/messense/trust-dns.git", branch = "tokio-1" }
trust-dns-resolver = { git = "https://github.com/messense/trust-dns.git", branch = "tokio-1" }

2
actix-codec/Cargo.toml
View File

@ -21,6 +21,6 @@ bytes = "1"
futures-core = { version = "0.3.7", default-features = false }
futures-sink = { version = "0.3.7", default-features = false }
log = "0.4"
pin-project = "1.0.0"
pin-project-lite = "0.2"
tokio = "1"
tokio-util = { version = "0.6", features = ["codec", "io"] }

34
actix-codec/src/framed.rs
View File

@ -5,7 +5,6 @@ use std::{fmt, io};
use bytes::{Buf, BytesMut};
use futures_core::{ready, Stream};
use futures_sink::Sink;
use pin_project::pin_project;
use crate::{AsyncRead, AsyncWrite, Decoder, Encoder};
@ -21,22 +20,23 @@ bitflags::bitflags! {
}
}
/// A unified `Stream` and `Sink` interface to an underlying I/O object, using
/// the `Encoder` and `Decoder` traits to encode and decode frames.
///
/// Raw I/O objects work with byte sequences, but higher-level code usually
/// wants to batch these into meaningful chunks, called "frames". This
/// method layers framing on top of an I/O object, by using the `Encoder`/`Decoder`
/// traits to handle encoding and decoding of message frames. Note that
/// the incoming and outgoing frame types may be distinct.
#[pin_project]
pub struct Framed<T, U> {
#[pin]
io: T,
codec: U,
flags: Flags,
read_buf: BytesMut,
write_buf: BytesMut,
pin_project_lite::pin_project! {
/// A unified `Stream` and `Sink` interface to an underlying I/O object, using
/// the `Encoder` and `Decoder` traits to encode and decode frames.
///
/// Raw I/O objects work with byte sequences, but higher-level code usually
/// wants to batch these into meaningful chunks, called "frames". This
/// method layers framing on top of an I/O object, by using the `Encoder`/`Decoder`
/// traits to handle encoding and decoding of message frames. Note that
/// the incoming and outgoing frame types may be distinct.
pub struct Framed<T, U> {
#[pin]
io: T,
codec: U,
flags: Flags,
read_buf: BytesMut,
write_buf: BytesMut,
}
}
impl<T, U> Framed<T, U>

6
actix-connect/Cargo.toml
View File

@ -33,7 +33,7 @@ uri = ["http"]
[dependencies]
actix-service = "1.0.6"
actix-codec = "0.3.0"
actix-utils = "3.0.0"
actix-utils = "2.0.0"
actix-rt = "1.1.1"
derive_more = "0.99.2"
@ -43,8 +43,8 @@ futures-util = { version = "0.3.7", default-features = false }
http = { version = "0.2.2", optional = true }
log = "0.4"
# FIXME: Use release version
trust-dns-proto = { git = "https://github.com/bluejekyll/trust-dns", branch = "main" }
trust-dns-resolver = { git = "https://github.com/bluejekyll/trust-dns", branch = "main" }
trust-dns-proto = "0.20.0-alpha.3"
trust-dns-resolver = "0.20.0-alpha.3"
# openssl
open-ssl = { package = "openssl", version = "0.10", optional = true }

4
actix-connect/src/connector.rs
View File

@ -74,9 +74,7 @@ impl<T: Address> Service<Connect<T>> for TcpConnector<T> {
type Error = ConnectError;
type Future = TcpConnectorResponse<T>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
actix_service::always_ready!();
fn call(&mut self, req: Connect<T>) -> Self::Future {
let port = req.port();

4
actix-connect/src/resolve.rs
View File

@ -106,9 +106,7 @@ impl<T: Address> Service<Connect<T>> for Resolver<T> {
type Error = ConnectError;
type Future = ResolverServiceFuture<T>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
actix_service::always_ready!();
fn call(&mut self, mut req: Connect<T>) -> Self::Future {
if req.addr.is_some() {

8
actix-connect/src/service.rs
View File

@ -94,9 +94,7 @@ impl<T: Address> Service<Connect<T>> for ConnectService<T> {
type Error = ConnectError;
type Future = ConnectServiceResponse<T>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
actix_service::always_ready!();
fn call(&mut self, req: Connect<T>) -> Self::Future {
ConnectServiceResponse {
@ -163,9 +161,7 @@ impl<T: Address + 'static> Service<Connect<T>> for TcpConnectService<T> {
type Error = ConnectError;
type Future = TcpConnectServiceResponse<T>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
actix_service::always_ready!();
fn call(&mut self, req: Connect<T>) -> Self::Future {
TcpConnectServiceResponse {

85
actix-connect/src/ssl/openssl.rs
View File

@ -1,5 +1,4 @@
use std::future::Future;
use std::marker::PhantomData;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::{fmt, io};
@ -19,90 +18,71 @@ use crate::{
};
/// OpenSSL connector factory
pub struct OpensslConnector<T, U> {
pub struct OpensslConnector {
connector: SslConnector,
_t: PhantomData<(T, U)>,
}
impl<T, U> OpensslConnector<T, U> {
impl OpensslConnector {
pub fn new(connector: SslConnector) -> Self {
OpensslConnector {
connector,
_t: PhantomData,
}
OpensslConnector { connector }
}
}
impl<T, U> OpensslConnector<T, U>
where
T: Address + 'static,
U: AsyncRead + AsyncWrite + Unpin + fmt::Debug + 'static,
{
pub fn service(connector: SslConnector) -> OpensslConnectorService<T, U> {
OpensslConnectorService {
connector,
_t: PhantomData,
}
impl OpensslConnector {
pub fn service(connector: SslConnector) -> OpensslConnectorService {
OpensslConnectorService { connector }
}
}
impl<T, U> Clone for OpensslConnector<T, U> {
impl Clone for OpensslConnector {
fn clone(&self) -> Self {
Self {
connector: self.connector.clone(),
_t: PhantomData,
}
}
}
impl<T, U> ServiceFactory for OpensslConnector<T, U>
impl<T, U> ServiceFactory<Connection<T, U>> for OpensslConnector
where
T: Address + 'static,
U: AsyncRead + AsyncWrite + Unpin + fmt::Debug + 'static,
{
type Request = Connection<T, U>;
type Response = Connection<T, SslStream<U>>;
type Error = io::Error;
type Config = ();
type Service = OpensslConnectorService<T, U>;
type Service = OpensslConnectorService;
type InitError = ();
type Future = Ready<Result<Self::Service, Self::InitError>>;
fn new_service(&self, _: ()) -> Self::Future {
ready(Ok(OpensslConnectorService {
connector: self.connector.clone(),
_t: PhantomData,
}))
}
}
pub struct OpensslConnectorService<T, U> {
pub struct OpensslConnectorService {
connector: SslConnector,
_t: PhantomData<(T, U)>,
}
impl<T, U> Clone for OpensslConnectorService<T, U> {
impl Clone for OpensslConnectorService {
fn clone(&self) -> Self {
Self {
connector: self.connector.clone(),
_t: PhantomData,
}
}
}
impl<T, U> Service for OpensslConnectorService<T, U>
impl<T, U> Service<Connection<T, U>> for OpensslConnectorService
where
T: Address + 'static,
U: AsyncRead + AsyncWrite + Unpin + fmt::Debug + 'static,
{
type Request = Connection<T, U>;
type Response = Connection<T, SslStream<U>>;
type Error = io::Error;
type Future = OpensslConnectorServiceFuture<T, U>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
actix_service::always_ready!();
fn call(&mut self, stream: Connection<T, U>) -> Self::Future {
match self.ssl_stream(stream) {
@ -112,18 +92,18 @@ where
}
}
impl<T, U> OpensslConnectorService<T, U>
where
T: Address + 'static,
U: AsyncRead + AsyncWrite + Unpin + fmt::Debug + 'static,
{
impl OpensslConnectorService {
// construct SslStream with connector.
// At this point SslStream does not perform any I/O.
// handshake would happen later in OpensslConnectorServiceFuture
fn ssl_stream(
fn ssl_stream<T, U>(
&self,
stream: Connection<T, U>,
) -> Result<(SslStream<U>, Connection<T, ()>), SslError> {
) -> Result<(SslStream<U>, Connection<T, ()>), SslError>
where
T: Address + 'static,
U: AsyncRead + AsyncWrite + Unpin + fmt::Debug + 'static,
{
trace!("SSL Handshake start for: {:?}", stream.host());
let (stream, connection) = stream.replace(());
let host = connection.host().to_string();
@ -176,7 +156,7 @@ where
pub struct OpensslConnectServiceFactory<T> {
tcp: ConnectServiceFactory<T>,
openssl: OpensslConnector<T, TcpStream>,
openssl: OpensslConnector,
}
impl<T> OpensslConnectServiceFactory<T> {
@ -202,7 +182,6 @@ impl<T> OpensslConnectServiceFactory<T> {
tcp: self.tcp.service(),
openssl: OpensslConnectorService {
connector: self.openssl.connector.clone(),
_t: PhantomData,
},
}
}
@ -217,8 +196,7 @@ impl<T> Clone for OpensslConnectServiceFactory<T> {
}
}
impl<T: Address + 'static> ServiceFactory for OpensslConnectServiceFactory<T> {
type Request = Connect<T>;
impl<T: Address + 'static> ServiceFactory<Connect<T>> for OpensslConnectServiceFactory<T> {
type Response = SslStream<TcpStream>;
type Error = ConnectError;
type Config = ();
@ -234,18 +212,15 @@ impl<T: Address + 'static> ServiceFactory for OpensslConnectServiceFactory<T> {
#[derive(Clone)]
pub struct OpensslConnectService<T> {
tcp: ConnectService<T>,
openssl: OpensslConnectorService<T, TcpStream>,
openssl: OpensslConnectorService,
}
impl<T: Address + 'static> Service for OpensslConnectService<T> {
type Request = Connect<T>;
impl<T: Address + 'static> Service<Connect<T>> for OpensslConnectService<T> {
type Response = SslStream<TcpStream>;
type Error = ConnectError;
type Future = OpensslConnectServiceResponse<T>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
actix_service::always_ready!();
fn call(&mut self, req: Connect<T>) -> Self::Future {
OpensslConnectServiceResponse {
@ -257,9 +232,9 @@ impl<T: Address + 'static> Service for OpensslConnectService<T> {
}
pub struct OpensslConnectServiceResponse<T: Address + 'static> {
fut1: Option<<ConnectService<T> as Service>::Future>,
fut2: Option<<OpensslConnectorService<T, TcpStream> as Service>::Future>,
openssl: OpensslConnectorService<T, TcpStream>,
fut1: Option<<ConnectService<T> as Service<Connect<T>>>::Future>,
fut2: Option<<OpensslConnectorService as Service<Connection<T, TcpStream>>>::Future>,
openssl: OpensslConnectorService,
}
impl<T: Address> Future for OpensslConnectServiceResponse<T> {
@ -267,7 +242,7 @@ impl<T: Address> Future for OpensslConnectServiceResponse<T> {
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
if let Some(ref mut fut) = self.fut1 {
match futures_util::ready!(Pin::new(fut).poll(cx)) {
match ready!(Pin::new(fut).poll(cx)) {
Ok(res) => {
let _ = self.fut1.take();
self.fut2 = Some(self.openssl.call(res));
@ -277,7 +252,7 @@ impl<T: Address> Future for OpensslConnectServiceResponse<T> {
}
if let Some(ref mut fut) = self.fut2 {
match futures_util::ready!(Pin::new(fut).poll(cx)) {
match ready!(Pin::new(fut).poll(cx)) {
Ok(connect) => Poll::Ready(Ok(connect.into_parts().0)),
Err(e) => Poll::Ready(Err(ConnectError::Io(io::Error::new(
io::ErrorKind::Other,

47
actix-connect/src/ssl/rustls.rs
View File

@ -1,6 +1,5 @@
use std::fmt;
use std::future::Future;
use std::marker::PhantomData;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
@ -18,88 +17,70 @@ use webpki::DNSNameRef;
use crate::{Address, Connection};
/// Rustls connector factory
pub struct RustlsConnector<T, U> {
pub struct RustlsConnector {
connector: Arc<ClientConfig>,
_t: PhantomData<(T, U)>,
}
impl<T, U> RustlsConnector<T, U> {
impl RustlsConnector {
pub fn new(connector: Arc<ClientConfig>) -> Self {
RustlsConnector {
connector,
_t: PhantomData,
}
RustlsConnector { connector }
}
}
impl<T, U> RustlsConnector<T, U>
where
T: Address,
U: AsyncRead + AsyncWrite + Unpin + fmt::Debug,
{
pub fn service(connector: Arc<ClientConfig>) -> RustlsConnectorService<T, U> {
RustlsConnectorService {
connector,
_t: PhantomData,
}
impl RustlsConnector {
pub fn service(connector: Arc<ClientConfig>) -> RustlsConnectorService {
RustlsConnectorService { connector }
}
}
impl<T, U> Clone for RustlsConnector<T, U> {
impl Clone for RustlsConnector {
fn clone(&self) -> Self {
Self {
connector: self.connector.clone(),
_t: PhantomData,
}
}
}
impl<T: Address, U> ServiceFactory for RustlsConnector<T, U>
impl<T: Address, U> ServiceFactory<Connection<T, U>> for RustlsConnector
where
U: AsyncRead + AsyncWrite + Unpin + fmt::Debug,
{
type Request = Connection<T, U>;
type Response = Connection<T, TlsStream<U>>;
type Error = std::io::Error;
type Config = ();
type Service = RustlsConnectorService<T, U>;
type Service = RustlsConnectorService;
type InitError = ();
type Future = Ready<Result<Self::Service, Self::InitError>>;
fn new_service(&self, _: ()) -> Self::Future {
ready(Ok(RustlsConnectorService {
connector: self.connector.clone(),
_t: PhantomData,
}))
}
}
pub struct RustlsConnectorService<T, U> {
pub struct RustlsConnectorService {
connector: Arc<ClientConfig>,
_t: PhantomData<(T, U)>,
}
impl<T, U> Clone for RustlsConnectorService<T, U> {
impl Clone for RustlsConnectorService {
fn clone(&self) -> Self {
Self {
connector: self.connector.clone(),
_t: PhantomData,
}
}
}
impl<T: Address, U> Service for RustlsConnectorService<T, U>
impl<T, U> Service<Connection<T, U>> for RustlsConnectorService
where
T: Address,
U: AsyncRead + AsyncWrite + Unpin + fmt::Debug,
{
type Request = Connection<T, U>;
type Response = Connection<T, TlsStream<U>>;
type Error = std::io::Error;
type Future = ConnectAsyncExt<T, U>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
actix_service::always_ready!();
fn call(&mut self, stream: Connection<T, U>) -> Self::Future {
trace!("SSL Handshake start for: {:?}", stream.host());

1
actix-rt/CHANGES.md
View File

@ -10,6 +10,7 @@
* Update `tokio` dependency to `1`
* Rename `time` module `delay_for` to `sleep`, `delay_until` to `sleep_until`, `Delay` to `Sleep` to keep inline with tokio.
* Remove `'static` lifetime requirement for `Runtime::block_on` and `SystemRunner::block_on`.
These methods would accept a &Self when calling.
Remove `'static` lifetime requirement for `System::run` and `Builder::run`.
`Arbiter::spawn` would panic when `System` is not in scope. [#207]

8
actix-server/Cargo.toml
View File

@ -23,19 +23,19 @@ default = []
actix-service = "1.0.6"
actix-rt = "1.1.1"
actix-codec = "0.3.0"
actix-utils = "3.0.0"
actix-utils = "2.0.0"
concurrent-queue = "1.2.2"
futures-channel = { version = "0.3.7", default-features = false }
futures-util = { version = "0.3.7", default-features = false }
futures-core = { version = "0.3.7", default-features = false }
log = "0.4"
mio = { version = "0.7.3", features = [ "os-poll", "tcp", "uds"] }
num_cpus = "1.13"
slab = "0.4"
tokio = { version = "1", features = ["sync"] }
[dev-dependencies]
actix-testing = "1.0.0"
bytes = "1"
env_logger = "0.7"
futures-util = { version = "0.3.7", default-features = false, features = ["sink"] }
tokio = { version = "1", features = ["full"] }
tokio = { version = "1", features = ["io-util"] }

2
actix-server/src/accept.rs
View File

@ -40,7 +40,7 @@ pub(crate) struct AcceptLoop {
impl AcceptLoop {
pub fn new(srv: Server) -> Self {
let poll = Poll::new().unwrap_or_else(|e| panic!("Can not create `mio::Poll`: {}", e));
let waker = WakerQueue::with_capacity(poll.registry(), 128)
let waker = WakerQueue::new(poll.registry())
.unwrap_or_else(|e| panic!("Can not create `mio::Waker`: {}", e));
Self {

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

@ -7,11 +7,9 @@ use std::{io, mem};
use actix_rt::net::TcpStream;
use actix_rt::time::{sleep_until, Instant};
use actix_rt::{spawn, System};
use futures_channel::mpsc::{unbounded, UnboundedReceiver};
use futures_channel::oneshot;
use futures_util::future::join_all;
use futures_util::stream::Stream;
use log::{error, info};
use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver};
use tokio::sync::oneshot;
use crate::accept::AcceptLoop;
use crate::config::{ConfiguredService, ServiceConfig};
@ -22,7 +20,7 @@ use crate::socket::{MioListener, StdSocketAddr, StdTcpListener, ToSocketAddrs};
use crate::socket::{MioTcpListener, MioTcpSocket};
use crate::waker_queue::{WakerInterest, WakerQueue};
use crate::worker::{self, Worker, WorkerAvailability, WorkerHandle};
use crate::Token;
use crate::{join_all, Token};
/// Server builder
pub struct ServerBuilder {
@ -50,7 +48,7 @@ impl Default for ServerBuilder {
impl ServerBuilder {
/// Create new Server builder instance
pub fn new() -> ServerBuilder {
let (tx, rx) = unbounded();
let (tx, rx) = unbounded_channel();
let server = Server::new(tx);
ServerBuilder {
@ -366,7 +364,8 @@ impl ServerBuilder {
let iter = self
.handles
.iter()
.map(move |worker| worker.1.stop(graceful));
.map(move |worker| worker.1.stop(graceful))
.collect();
let fut = join_all(iter);
@ -439,7 +438,7 @@ impl Future for ServerBuilder {
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
loop {
match Pin::new(&mut self.cmd).poll_next(cx) {
match Pin::new(&mut self.cmd).poll_recv(cx) {
Poll::Ready(Some(it)) => self.as_mut().get_mut().handle_cmd(it),
_ => return Poll::Pending,
}

5
actix-server/src/config.rs
View File

@ -8,14 +8,13 @@ use actix_service::{
ServiceFactory as BaseServiceFactory,
};
use actix_utils::counter::CounterGuard;
use futures_util::future::ready;
use futures_core::future::LocalBoxFuture;
use log::error;
use crate::builder::bind_addr;
use crate::service::{BoxedServerService, InternalServiceFactory, StreamService};
use crate::socket::{MioStream, MioTcpListener, StdSocketAddr, StdTcpListener, ToSocketAddrs};
use crate::LocalBoxFuture;
use crate::Token;
use crate::{ready, Token};
pub struct ServiceConfig {
pub(crate) services: Vec<(String, MioTcpListener)>,

80
actix-server/src/lib.rs
View File

@ -22,6 +22,10 @@ pub use self::service::ServiceFactory;
#[doc(hidden)]
pub use self::socket::FromStream;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
/// Socket ID token
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub(crate) struct Token(usize);
@ -44,10 +48,80 @@ impl Token {
}
}
pub(crate) type LocalBoxFuture<'a, T> =
std::pin::Pin<Box<dyn std::future::Future<Output = T> + 'a>>;
/// Start server building process
pub fn new() -> ServerBuilder {
ServerBuilder::default()
}
// temporary Ready type for std::future::{ready, Ready}; Can be removed when MSRV surpass 1.48
#[doc(hidden)]
pub struct Ready<T>(Option<T>);
pub(crate) fn ready<T>(t: T) -> Ready<T> {
Ready(Some(t))
}
impl<T> Unpin for Ready<T> {}
impl<T> Future for Ready<T> {
type Output = T;
fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Self::Output> {
Poll::Ready(self.get_mut().0.take().unwrap())
}
}
// a poor man's join future. joined future is only used when starting/stopping the server.
// pin_project and pinned futures are overkill for this task.
pub(crate) struct JoinAll<T> {
fut: Vec<JoinFuture<T>>,
}
pub(crate) fn join_all<T>(fut: Vec<impl Future<Output = T> + 'static>) -> JoinAll<T> {
let fut = fut
.into_iter()
.map(|f| JoinFuture::Future(Box::pin(f)))
.collect();
JoinAll { fut }
}
enum JoinFuture<T> {
Future(Pin<Box<dyn Future<Output = T>>>),
Result(Option<T>),
}
impl<T> Unpin for JoinAll<T> {}
impl<T> Future for JoinAll<T> {
type Output = Vec<T>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut ready = true;
let this = self.get_mut();
for fut in this.fut.iter_mut() {
if let JoinFuture::Future(f) = fut {
match f.as_mut().poll(cx) {
Poll::Ready(t) => {
*fut = JoinFuture::Result(Some(t));
}
Poll::Pending => ready = false,
}
}
}
if ready {
let mut res = Vec::new();
for fut in this.fut.iter_mut() {
if let JoinFuture::Result(f) = fut {
res.push(f.take().unwrap());
}
}
Poll::Ready(res)
} else {
Poll::Pending
}
}
}

16
actix-server/src/server.rs
View File

@ -3,8 +3,8 @@ use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};
use futures_channel::mpsc::UnboundedSender;
use futures_channel::oneshot;
use tokio::sync::mpsc::UnboundedSender;
use tokio::sync::oneshot;
use crate::builder::ServerBuilder;
use crate::signals::Signal;
@ -41,11 +41,11 @@ impl Server {
}
pub(crate) fn signal(&self, sig: Signal) {
let _ = self.0.unbounded_send(ServerCommand::Signal(sig));
let _ = self.0.send(ServerCommand::Signal(sig));
}
pub(crate) fn worker_faulted(&self, idx: usize) {
let _ = self.0.unbounded_send(ServerCommand::WorkerFaulted(idx));
let _ = self.0.send(ServerCommand::WorkerFaulted(idx));
}
/// Pause accepting incoming connections
@ -54,7 +54,7 @@ impl Server {
/// All opened connection remains active.
pub fn pause(&self) -> impl Future<Output = ()> {
let (tx, rx) = oneshot::channel();
let _ = self.0.unbounded_send(ServerCommand::Pause(tx));
let _ = self.0.send(ServerCommand::Pause(tx));
async {
let _ = rx.await;
}
@ -63,7 +63,7 @@ impl Server {
/// Resume accepting incoming connections
pub fn resume(&self) -> impl Future<Output = ()> {
let (tx, rx) = oneshot::channel();
let _ = self.0.unbounded_send(ServerCommand::Resume(tx));
let _ = self.0.send(ServerCommand::Resume(tx));
async {
let _ = rx.await;
}
@ -74,7 +74,7 @@ impl Server {
/// If server starts with `spawn()` method, then spawned thread get terminated.
pub fn stop(&self, graceful: bool) -> impl Future<Output = ()> {
let (tx, rx) = oneshot::channel();
let _ = self.0.unbounded_send(ServerCommand::Stop {
let _ = self.0.send(ServerCommand::Stop {
graceful,
completion: Some(tx),
});
@ -98,7 +98,7 @@ impl Future for Server {
if this.1.is_none() {
let (tx, rx) = oneshot::channel();
if this.0.unbounded_send(ServerCommand::Notify(tx)).is_err() {
if this.0.send(ServerCommand::Notify(tx)).is_err() {
return Poll::Ready(Ok(()));
}
this.1 = Some(rx);

6
actix-server/src/service.rs
View File

@ -2,15 +2,13 @@ use std::marker::PhantomData;
use std::net::SocketAddr;
use std::task::{Context, Poll};
use actix_rt::spawn;
use actix_service::{Service, ServiceFactory as BaseServiceFactory};
use actix_utils::counter::CounterGuard;
use futures_util::future::{ready, Ready};
use futures_core::future::LocalBoxFuture;
use log::error;
use crate::socket::{FromStream, MioStream};
use crate::LocalBoxFuture;
use crate::Token;
use crate::{ready, Ready, Token};
pub trait ServiceFactory<Stream: FromStream>: Send + Clone + 'static {
type Factory: BaseServiceFactory<Stream, Config = ()>;

3
actix-server/src/signals.rs
View File

@ -2,8 +2,9 @@ use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use futures_core::future::LocalBoxFuture;
use crate::server::Server;
use crate::LocalBoxFuture;
/// Different types of process signals
#[allow(dead_code)]

7
actix-server/src/waker_queue.rs
View File

@ -32,9 +32,9 @@ impl WakerQueue {
///
/// A fixed `WAKER_TOKEN` is used to identify the wake interest and the `Poll` needs to match
/// event's token for it to properly handle `WakerInterest`.
pub(crate) fn with_capacity(registry: &Registry, cap: usize) -> std::io::Result<Self> {
pub(crate) fn new(registry: &Registry) -> std::io::Result<Self> {
let waker = Waker::new(registry, WAKER_TOKEN)?;
let queue = ConcurrentQueue::bounded(cap);
let queue = ConcurrentQueue::unbounded();
Ok(Self(Arc::new((waker, queue))))
}
@ -43,10 +43,9 @@ impl WakerQueue {
pub(crate) fn wake(&self, interest: WakerInterest) {
let (waker, queue) = self.deref();
// FIXME: should we handle error here?
queue
.push(interest)
.unwrap_or_else(|e| panic!("WakerQueue overflow: {}", e));
.unwrap_or_else(|e| panic!("WakerQueue closed: {}", e));
waker
.wake()

34
actix-server/src/worker.rs
View File

@ -8,18 +8,15 @@ use std::time::Duration;
use actix_rt::time::{sleep_until, Instant, Sleep};
use actix_rt::{spawn, Arbiter};
use actix_utils::counter::Counter;
use futures_channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender};
use futures_channel::oneshot;
use futures_util::future::join_all;
use futures_util::stream::Stream;
use futures_util::TryFutureExt;
use futures_core::future::LocalBoxFuture;
use log::{error, info, trace};
use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
use tokio::sync::oneshot;
use crate::service::{BoxedServerService, InternalServiceFactory};
use crate::socket::{MioStream, SocketAddr};
use crate::waker_queue::{WakerInterest, WakerQueue};
use crate::LocalBoxFuture;
use crate::Token;
use crate::{join_all, Token};
pub(crate) struct WorkerCommand(Conn);
@ -84,9 +81,7 @@ impl WorkerHandle {
}
pub fn send(&self, msg: Conn) -> Result<(), Conn> {
self.tx1
.unbounded_send(WorkerCommand(msg))
.map_err(|msg| msg.into_inner().0)
self.tx1.send(WorkerCommand(msg)).map_err(|msg| msg.0 .0)
}
pub fn available(&self) -> bool {
@ -95,7 +90,7 @@ impl WorkerHandle {
pub fn stop(&self, graceful: bool) -> oneshot::Receiver<bool> {
let (result, rx) = oneshot::channel();
let _ = self.tx2.unbounded_send(StopCommand { graceful, result });
let _ = self.tx2.send(StopCommand { graceful, result });
rx
}
}
@ -172,8 +167,8 @@ impl Worker {
availability: WorkerAvailability,
shutdown_timeout: Duration,
) -> WorkerHandle {
let (tx1, rx) = unbounded();
let (tx2, rx2) = unbounded();
let (tx1, rx) = unbounded_channel();
let (tx2, rx2) = unbounded_channel();
let avail = availability.clone();
// every worker runs in it's own arbiter.
@ -195,9 +190,12 @@ impl Worker {
.iter()
.enumerate()
.map(|(idx, factory)| {
factory.create().map_ok(move |r| {
r.into_iter().map(|(t, s)| (idx, t, s)).collect::<Vec<_>>()
})
let fut = factory.create();
async move {
fut.await.map(|r| {
r.into_iter().map(|(t, s)| (idx, t, s)).collect::<Vec<_>>()
})
}
})
.collect::<Vec<_>>();
@ -312,7 +310,7 @@ impl Future for Worker {
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
// `StopWorker` message handler
if let Poll::Ready(Some(StopCommand { graceful, result })) =
Pin::new(&mut self.rx2).poll_next(cx)
Pin::new(&mut self.rx2).poll_recv(cx)
{
self.availability.set(false);
let num = num_connections();
@ -432,7 +430,7 @@ impl Future for Worker {
}
}
match Pin::new(&mut self.rx).poll_next(cx) {
match Pin::new(&mut self.rx).poll_recv(cx) {
// handle incoming io stream
Poll::Ready(Some(WorkerCommand(msg))) => {
let guard = self.conns.get();

4
actix-server/tests/test_server.rs
View File

@ -49,14 +49,14 @@ fn test_listen() {
let h = thread::spawn(move || {
let sys = actix_rt::System::new("test");
let lst = net::TcpListener::bind(addr).unwrap();
sys.block_on(lazy(|_| {
sys.block_on(async {
Server::build()
.disable_signals()
.workers(1)
.listen("test", lst, move || fn_service(|_| ok::<_, ()>(())))
.unwrap()
.start()
}));
});
let _ = tx.send(actix_rt::System::current());
let _ = sys.run();
});

10
actix-service/CHANGES.md
View File

@ -3,10 +3,16 @@
## Unreleased - 2020-xx-xx
* `Service`, other traits, and many type signatures now take the the request type as a type
parameter instead of an associated type. [#232]
* Upgrade `pin-project` to `1.0`.
* Add `always_ready!` and `forward_ready!` macros. [#233]
* Crate is now `no_std`. [#233]
* Migrate pin projections to `pin-project-lite`. [#233]
* Remove `AndThenApplyFn` and Pipeline `and_then_apply_fn`. Use the
`.and_then(apply_fn(...))` construction. [#233]
* Move non-vital methods to `ServiceExt` and `ServiceFactoryExt` extension traits. [#235]
[#232]: https://github.com/actix/actix-net/pull/232
[#233]: https://github.com/actix/actix-net/pull/233
[#235]: https://github.com/actix/actix-net/pull/235
## 1.0.6 - 2020-08-09

14
actix-service/Cargo.toml
View File

@ -17,17 +17,9 @@ name = "actix_service"
path = "src/lib.rs"
[dependencies]
futures-util = "0.3.1"
pin-project = "1.0.0"
futures-core = { version = "0.3.7", default-features = false }
pin-project-lite = "0.2"
[dev-dependencies]
actix-rt = "1.0.0"
criterion = "0.3"
[[bench]]
name = "unsafecell_vs_refcell"
harness = false
[[bench]]
name = "and_then"
harness = false
futures-util = { version = "0.3.7", default-features = false }

334
actix-service/benches/and_then.rs
View File

@ -1,334 +0,0 @@
use actix_service::boxed::BoxFuture;
use actix_service::IntoService;
use actix_service::Service;
/// Benchmark various implementations of and_then
use criterion::{criterion_main, Criterion};
use futures_util::future::join_all;
use futures_util::future::TryFutureExt;
use std::future::Future;
use std::pin::Pin;
use std::rc::Rc;
use std::task::{Context, Poll};
use std::{
cell::{RefCell, UnsafeCell},
marker::PhantomData,
};
/*
* Test services A,B for AndThen service implementations
*/
async fn svc1(_: ()) -> Result<usize, ()> {
Ok(1)
}
async fn svc2(req: usize) -> Result<usize, ()> {
Ok(req + 1)
}
/*
* AndThenUC - original AndThen service based on UnsafeCell
* Cut down version of actix_service::AndThenService based on actix-service::Cell
*/
struct AndThenUC<A, Req, B>(Rc<UnsafeCell<(A, B)>>, PhantomData<Req>);
impl<A, Req, B> AndThenUC<A, Req, B> {
fn new(a: A, b: B) -> Self
where
A: Service<Req>,
B: Service<A::Response, Error = A::Error>,
{
Self(Rc::new(UnsafeCell::new((a, b))), PhantomData)
}
}
impl<A, Req, B> Clone for AndThenUC<A, Req, B> {
fn clone(&self) -> Self {
Self(self.0.clone(), PhantomData)
}
}
impl<A, Req, B> Service<Req> for AndThenUC<A, Req, B>
where
A: Service<Req>,
B: Service<A::Response, Error = A::Error>,
{
type Response = B::Response;
type Error = A::Error;
type Future = AndThenServiceResponse<A, Req, B>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, req: Req) -> Self::Future {
let fut = unsafe { &mut *(*self.0).get() }.0.call(req);
AndThenServiceResponse {
state: State::A(fut, Some(self.0.clone())),
_phantom: PhantomData,
}
}
}
#[pin_project::pin_project]
pub(crate) struct AndThenServiceResponse<A, Req, B>
where
A: Service<Req>,
B: Service<A::Response, Error = A::Error>,
{
#[pin]
state: State<A, Req, B>,
_phantom: PhantomData<Req>,
}
#[pin_project::pin_project(project = StateProj)]
enum State<A, Req, B>
where
A: Service<Req>,
B: Service<A::Response, Error = A::Error>,
{
A(#[pin] A::Future, Option<Rc<UnsafeCell<(A, B)>>>),
B(#[pin] B::Future),
Empty(PhantomData<Req>),
}
impl<A, Req, B> Future for AndThenServiceResponse<A, Req, B>
where
A: Service<Req>,
B: Service<A::Response, Error = A::Error>,
{
type Output = Result<B::Response, A::Error>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.as_mut().project();
match this.state.as_mut().project() {
StateProj::A(fut, b) => match fut.poll(cx)? {
Poll::Ready(res) => {
let b = b.take().unwrap();
this.state.set(State::Empty(PhantomData)); // drop fut A
let fut = unsafe { &mut (*b.get()).1 }.call(res);
this.state.set(State::B(fut));
self.poll(cx)
}
Poll::Pending => Poll::Pending,
},
StateProj::B(fut) => fut.poll(cx).map(|r| {
this.state.set(State::Empty(PhantomData));
r
}),
StateProj::Empty(_) => {
panic!("future must not be polled after it returned `Poll::Ready`")
}
}
}
}
/*
* AndThenRC - AndThen service based on RefCell
*/
struct AndThenRC<A, Req, B>(Rc<RefCell<(A, B)>>, PhantomData<Req>);
impl<A, Req, B> AndThenRC<A, Req, B> {
fn new(a: A, b: B) -> Self
where
A: Service<Req>,
B: Service<A::Response, Error = A::Error>,
{
Self(Rc::new(RefCell::new((a, b))), PhantomData)
}
}
impl<A, Req, B> Clone for AndThenRC<A, Req, B> {
fn clone(&self) -> Self {
Self(self.0.clone(), PhantomData)
}
}
impl<A, Req, B> Service<Req> for AndThenRC<A, Req, B>
where
A: Service<Req>,
B: Service<A::Response, Error = A::Error>,
{
type Response = B::Response;
type Error = A::Error;
type Future = AndThenServiceResponseRC<A, Req, B>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, req: Req) -> Self::Future {
let fut = self.0.borrow_mut().0.call(req);
AndThenServiceResponseRC {
state: StateRC::A(fut, Some(self.0.clone())),
}
}
}
#[pin_project::pin_project]
pub(crate) struct AndThenServiceResponseRC<A, Req, B>
where
A: Service<Req>,
B: Service<A::Response, Error = A::Error>,
{
#[pin]
state: StateRC<A, Req, B>,
}
#[pin_project::pin_project(project = StateRCProj)]
enum StateRC<A, Req, B>
where
A: Service<Req>,
B: Service<A::Response, Error = A::Error>,
{
A(#[pin] A::Future, Option<Rc<RefCell<(A, B)>>>),
B(#[pin] B::Future),
Empty(PhantomData<Req>),
}
impl<A, Req, B> Future for AndThenServiceResponseRC<A, Req, B>
where
A: Service<Req>,
B: Service<A::Response, Error = A::Error>,
{
type Output = Result<B::Response, A::Error>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.as_mut().project();
match this.state.as_mut().project() {
StateRCProj::A(fut, b) => match fut.poll(cx)? {
Poll::Ready(res) => {
let b = b.take().unwrap();
this.state.set(StateRC::Empty(PhantomData)); // drop fut A
let fut = b.borrow_mut().1.call(res);
this.state.set(StateRC::B(fut));
self.poll(cx)
}
Poll::Pending => Poll::Pending,
},
StateRCProj::B(fut) => fut.poll(cx).map(|r| {
this.state.set(StateRC::Empty(PhantomData));
r
}),
StateRCProj::Empty(_) => {
panic!("future must not be polled after it returned `Poll::Ready`")
}
}
}
}
/*
* AndThenRCFuture - AndThen service based on RefCell
* and standard futures::future::and_then combinator in a Box
*/
struct AndThenRCFuture<A, Req, B>(Rc<RefCell<(A, B)>>, PhantomData<Req>);
impl<A, Req, B> AndThenRCFuture<A, Req, B> {
fn new(a: A, b: B) -> Self
where
A: Service<Req>,
B: Service<A::Response, Error = A::Error>,
{
Self(Rc::new(RefCell::new((a, b))), PhantomData)
}
}
impl<A, Req, B> Clone for AndThenRCFuture<A, Req, B> {
fn clone(&self) -> Self {
Self(self.0.clone(), PhantomData)
}
}
impl<A, Req, B> Service<Req> for AndThenRCFuture<A, Req, B>
where
A: Service<Req> + 'static,
A::Future: 'static,
B: Service<A::Response, Error = A::Error> + 'static,
B::Future: 'static,
{
type Response = B::Response;
type Error = A::Error;
type Future = BoxFuture<Self::Response, Self::Error>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, req: Req) -> Self::Future {
let fut = self.0.borrow_mut().0.call(req);
let core = self.0.clone();
let fut2 = move |res| (*core).borrow_mut().1.call(res);
Box::pin(fut.and_then(fut2))
}
}
/// Criterion Benchmark for async Service
/// Should be used from within criterion group:
/// ```rust,ignore
/// let mut criterion: ::criterion::Criterion<_> =
/// ::criterion::Criterion::default().configure_from_args();
/// bench_async_service(&mut criterion, ok_service(), "async_service_direct");
/// ```
///
/// Usable for benching Service wrappers:
/// Using minimum service code implementation we first measure
/// time to run minimum service, then measure time with wrapper.
///
/// Sample output
/// async_service_direct time: [1.0908 us 1.1656 us 1.2613 us]
pub fn bench_async_service<S>(c: &mut Criterion, srv: S, name: &str)
where
S: Service<(), Response = usize, Error = ()> + Clone + 'static,
{
let rt = actix_rt::System::new("test");
// start benchmark loops
c.bench_function(name, move |b| {
b.iter_custom(|iters| {
let mut srvs: Vec<_> = (1..iters).map(|_| srv.clone()).collect();
// exclude request generation, it appears it takes significant time vs call (3us vs 1us)
let start = std::time::Instant::now();
// benchmark body
rt.block_on(async move { join_all(srvs.iter_mut().map(|srv| srv.call(()))).await });
// check that at least first request succeeded
start.elapsed()
})
});
}
pub fn service_benches() {
let mut criterion: ::criterion::Criterion<_> =
::criterion::Criterion::default().configure_from_args();
bench_async_service(
&mut criterion,
AndThenUC::new(svc1.into_service(), svc2.into_service()),
"AndThen with UnsafeCell",
);
bench_async_service(
&mut criterion,
AndThenRC::new(svc1.into_service(), svc2.into_service()),
"AndThen with RefCell",
);
bench_async_service(
&mut criterion,
AndThenUC::new(svc1.into_service(), svc2.into_service()),
"AndThen with UnsafeCell",
);
bench_async_service(
&mut criterion,
AndThenRC::new(svc1.into_service(), svc2.into_service()),
"AndThen with RefCell",
);
bench_async_service(
&mut criterion,
AndThenRCFuture::new(svc1.into_service(), svc2.into_service()),
"AndThen with RefCell via future::and_then",
);
}
criterion_main!(service_benches);

110
actix-service/benches/unsafecell_vs_refcell.rs
View File

@ -1,110 +0,0 @@
use actix_service::Service;
use criterion::{criterion_main, Criterion};
use futures_util::future::join_all;
use futures_util::future::{ok, Ready};
use std::cell::{RefCell, UnsafeCell};
use std::rc::Rc;
use std::task::{Context, Poll};
struct SrvUC(Rc<UnsafeCell<usize>>);
impl Default for SrvUC {
fn default() -> Self {
Self(Rc::new(UnsafeCell::new(0)))
}
}
impl Clone for SrvUC {
fn clone(&self) -> Self {
Self(self.0.clone())
}
}
impl Service<()> for SrvUC {
type Response = usize;
type Error = ();
type Future = Ready<Result<Self::Response, ()>>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, _: ()) -> Self::Future {
unsafe { *(*self.0).get() = *(*self.0).get() + 1 };
ok(unsafe { *self.0.get() })
}
}
struct SrvRC(Rc<RefCell<usize>>);
impl Default for SrvRC {
fn default() -> Self {
Self(Rc::new(RefCell::new(0)))
}
}
impl Clone for SrvRC {
fn clone(&self) -> Self {
Self(self.0.clone())
}
}
impl Service<()> for SrvRC {
type Response = usize;
type Error = ();
type Future = Ready<Result<Self::Response, ()>>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, _: ()) -> Self::Future {
let prev = *self.0.borrow();
*(*self.0).borrow_mut() = prev + 1;
ok(*self.0.borrow())
}
}
/// Criterion Benchmark for async Service
/// Should be used from within criterion group:
/// ```rust,ignore
/// let mut criterion: ::criterion::Criterion<_> =
/// ::criterion::Criterion::default().configure_from_args();
/// bench_async_service(&mut criterion, ok_service(), "async_service_direct");
/// ```
///
/// Usable for benching Service wrappers:
/// Using minimum service code implementation we first measure
/// time to run minimum service, then measure time with wrapper.
///
/// Sample output
/// async_service_direct time: [1.0908 us 1.1656 us 1.2613 us]
pub fn bench_async_service<S>(c: &mut Criterion, srv: S, name: &str)
where
S: Service<(), Response = usize, Error = ()> + Clone + 'static,
{
let rt = actix_rt::System::new("test");
// start benchmark loops
c.bench_function(name, move |b| {
b.iter_custom(|iters| {
let mut srvs: Vec<_> = (1..iters).map(|_| srv.clone()).collect();
// exclude request generation, it appears it takes significant time vs call (3us vs 1us)
let start = std::time::Instant::now();
// benchmark body
rt.block_on(async { join_all(srvs.iter_mut().map(|srv| srv.call(()))).await });
// check that at least first request succeeded
start.elapsed()
})
});
}
pub fn service_benches() {
let mut criterion: ::criterion::Criterion<_> =
::criterion::Criterion::default().configure_from_args();
bench_async_service(&mut criterion, SrvUC::default(), "Service with UnsafeCell");
bench_async_service(&mut criterion, SrvRC::default(), "Service with RefCell");
bench_async_service(&mut criterion, SrvUC::default(), "Service with UnsafeCell");
bench_async_service(&mut criterion, SrvRC::default(), "Service with RefCell");
}
criterion_main!(service_benches);

109
actix-service/src/and_then.rs
View File

@ -1,8 +1,13 @@
use std::future::Future;
use std::pin::Pin;
use std::rc::Rc;
use std::task::{Context, Poll};
use std::{cell::RefCell, marker::PhantomData};
use alloc::rc::Rc;
use core::{
cell::RefCell,
future::Future,
marker::PhantomData,
pin::Pin,
task::{Context, Poll},
};
use pin_project_lite::pin_project;
use super::{Service, ServiceFactory};
@ -50,30 +55,43 @@ where
fn call(&mut self, req: Req) -> Self::Future {
AndThenServiceResponse {
state: State::A(self.0.borrow_mut().0.call(req), Some(self.0.clone())),
state: State::A {
fut: self.0.borrow_mut().0.call(req),
b: Some(self.0.clone()),
},
}
}
}
#[pin_project::pin_project]
pub(crate) struct AndThenServiceResponse<A, B, Req>
where
A: Service<Req>,
B: Service<A::Response, Error = A::Error>,
{
#[pin]
state: State<A, B, Req>,
pin_project! {
pub(crate) struct AndThenServiceResponse<A, B, Req>
where
A: Service<Req>,
B: Service<A::Response, Error = A::Error>,
{
#[pin]
state: State<A, B, Req>,
}
}
#[pin_project::pin_project(project = StateProj)]
enum State<A, B, Req>
where
A: Service<Req>,
B: Service<A::Response, Error = A::Error>,
{
A(#[pin] A::Future, Option<Rc<RefCell<(A, B)>>>),
B(#[pin] B::Future),
Empty,
pin_project! {
#[project = StateProj]
enum State<A, B, Req>
where
A: Service<Req>,
B: Service<A::Response, Error = A::Error>,
{
A {
#[pin]
fut: A::Future,
b: Option<Rc<RefCell<(A, B)>>>,
},
B {
#[pin]
fut: B::Future,
},
Empty,
}
}
impl<A, B, Req> Future for AndThenServiceResponse<A, B, Req>
@ -87,17 +105,17 @@ where
let mut this = self.as_mut().project();
match this.state.as_mut().project() {
StateProj::A(fut, b) => match fut.poll(cx)? {
StateProj::A { fut, b } => match fut.poll(cx)? {
Poll::Ready(res) => {
let b = b.take().unwrap();
this.state.set(State::Empty); // drop fut A
let fut = b.borrow_mut().1.call(res);
this.state.set(State::B(fut));
this.state.set(State::B { fut });
self.poll(cx)
}
Poll::Pending => Poll::Pending,
},
StateProj::B(fut) => fut.poll(cx).map(|r| {
StateProj::B { fut } => fut.poll(cx).map(|r| {
this.state.set(State::Empty);
r
}),
@ -191,19 +209,20 @@ where
}
}
#[pin_project::pin_project]
pub(crate) struct AndThenServiceFactoryResponse<A, B, Req>
where
A: ServiceFactory<Req>,
B: ServiceFactory<A::Response>,
{
#[pin]
fut_a: A::Future,
#[pin]
fut_b: B::Future,
pin_project! {
pub(crate) struct AndThenServiceFactoryResponse<A, B, Req>
where
A: ServiceFactory<Req>,
B: ServiceFactory<A::Response>,
{
#[pin]
fut_a: A::Future,
#[pin]
fut_b: B::Future,
a: Option<A::Service>,
b: Option<B::Service>,
a: Option<A::Service>,
b: Option<B::Service>,
}
}
impl<A, B, Req> AndThenServiceFactoryResponse<A, B, Req>
@ -254,13 +273,17 @@ where
#[cfg(test)]
mod tests {
use std::cell::Cell;
use std::rc::Rc;
use std::task::{Context, Poll};
use alloc::rc::Rc;
use core::{
cell::Cell,
task::{Context, Poll},
};
use futures_util::future::{lazy, ok, ready, Ready};
use futures_util::future::lazy;
use crate::{fn_factory, pipeline, pipeline_factory, Service, ServiceFactory};
use crate::{
fn_factory, ok, pipeline, pipeline_factory, ready, Ready, Service, ServiceFactory,
};
struct Srv1(Rc<Cell<usize>>);

334
actix-service/src/and_then_apply_fn.rs
View File

@ -1,334 +0,0 @@
use std::cell::RefCell;
use std::future::Future;
use std::marker::PhantomData;
use std::pin::Pin;
use std::rc::Rc;
use std::task::{Context, Poll};
use crate::{Service, ServiceFactory};
/// `Apply` service combinator
pub(crate) struct AndThenApplyFn<S1, S2, F, Fut, Req, In, Res, Err>
where
S1: Service<Req>,
S2: Service<In>,
F: FnMut(S1::Response, &mut S2) -> Fut,
Fut: Future<Output = Result<Res, Err>>,
Err: From<S1::Error> + From<S2::Error>,
{
svc: Rc<RefCell<(S1, S2, F)>>,
_phantom: PhantomData<(Fut, Req, In, Res, Err)>,
}
impl<S1, S2, F, Fut, Req, In, Res, Err> AndThenApplyFn<S1, S2, F, Fut, Req, In, Res, Err>
where
S1: Service<Req>,
S2: Service<In>,
F: FnMut(S1::Response, &mut S2) -> Fut,
Fut: Future<Output = Result<Res, Err>>,
Err: From<S1::Error> + From<S2::Error>,
{
/// Create new `Apply` combinator
pub(crate) fn new(a: S1, b: S2, wrap_fn: F) -> Self {
Self {
svc: Rc::new(RefCell::new((a, b, wrap_fn))),
_phantom: PhantomData,
}
}
}
impl<S1, S2, F, Fut, Req, In, Res, Err> Clone
for AndThenApplyFn<S1, S2, F, Fut, Req, In, Res, Err>
where
S1: Service<Req>,
S2: Service<In>,
F: FnMut(S1::Response, &mut S2) -> Fut,
Fut: Future<Output = Result<Res, Err>>,
Err: From<S1::Error> + From<S2::Error>,
{
fn clone(&self) -> Self {
AndThenApplyFn {
svc: self.svc.clone(),
_phantom: PhantomData,
}
}
}
impl<S1, S2, F, Fut, Req, In, Res, Err> Service<Req>
for AndThenApplyFn<S1, S2, F, Fut, Req, In, Res, Err>
where
S1: Service<Req>,
S2: Service<In>,
F: FnMut(S1::Response, &mut S2) -> Fut,
Fut: Future<Output = Result<Res, Err>>,
Err: From<S1::Error> + From<S2::Error>,
{
type Response = Res;
type Error = Err;
type Future = AndThenApplyFnFuture<S1, S2, F, Fut, Req, In, Res, Err>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
let mut inner = self.svc.borrow_mut();
let not_ready = inner.0.poll_ready(cx)?.is_pending();
if inner.1.poll_ready(cx)?.is_pending() || not_ready {
Poll::Pending
} else {
Poll::Ready(Ok(()))
}
}
fn call(&mut self, req: Req) -> Self::Future {
let fut = self.svc.borrow_mut().0.call(req);
AndThenApplyFnFuture {
state: State::A(fut, Some(self.svc.clone())),
}
}
}
#[pin_project::pin_project]
pub(crate) struct AndThenApplyFnFuture<S1, S2, F, Fut, Req, In, Res, Err>
where
S1: Service<Req>,
S2: Service<In>,
F: FnMut(S1::Response, &mut S2) -> Fut,
Fut: Future<Output = Result<Res, Err>>,
Err: From<S1::Error> + From<S2::Error>,
{
#[pin]
state: State<S1, S2, F, Fut, Req, In, Res, Err>,
}
#[pin_project::pin_project(project = StateProj)]
enum State<S1, S2, F, Fut, Req, In, Res, Err>
where
S1: Service<Req>,
S2: Service<In>,
F: FnMut(S1::Response, &mut S2) -> Fut,
Fut: Future<Output = Result<Res, Err>>,
Err: From<S1::Error> + From<S2::Error>,
{
A(#[pin] S1::Future, Option<Rc<RefCell<(S1, S2, F)>>>),
B(#[pin] Fut),
Empty(PhantomData<In>),
}
impl<S1, S2, F, Fut, Req, In, Res, Err> Future
for AndThenApplyFnFuture<S1, S2, F, Fut, Req, In, Res, Err>
where
S1: Service<Req>,
S2: Service<In>,
F: FnMut(S1::Response, &mut S2) -> Fut,
Fut: Future<Output = Result<Res, Err>>,
Err: From<S1::Error> + From<S2::Error>,
{
type Output = Result<Res, Err>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.as_mut().project();
match this.state.as_mut().project() {
StateProj::A(fut, b) => match fut.poll(cx)? {
Poll::Ready(res) => {
let b = Option::take(b).unwrap();
this.state.set(State::Empty(PhantomData));
let (_, b, f) = &mut *b.borrow_mut();
let fut = f(res, b);
this.state.set(State::B(fut));
self.poll(cx)
}
Poll::Pending => Poll::Pending,
},
StateProj::B(fut) => fut.poll(cx).map(|r| {
this.state.set(State::Empty(PhantomData));
r
}),
StateProj::Empty(_) => {
panic!("future must not be polled after it returned `Poll::Ready`")
}
}
}
}
/// `AndThenApplyFn` service factory
pub(crate) struct AndThenApplyFnFactory<SF1, SF2, F, Fut, Req, In, Res, Err> {
srv: Rc<(SF1, SF2, F)>,
_phantom: PhantomData<(Fut, Req, In, Res, Err)>,
}
impl<SF1, SF2, F, Fut, Req, In, Res, Err>
AndThenApplyFnFactory<SF1, SF2, F, Fut, Req, In, Res, Err>
where
SF1: ServiceFactory<Req>,
SF2: ServiceFactory<In, Config = SF1::Config, InitError = SF1::InitError>,
F: FnMut(SF1::Response, &mut SF2::Service) -> Fut + Clone,
Fut: Future<Output = Result<Res, Err>>,
Err: From<SF1::Error> + From<SF2::Error>,
{
/// Create new `ApplyNewService` new service instance
pub(crate) fn new(a: SF1, b: SF2, wrap_fn: F) -> Self {
Self {
srv: Rc::new((a, b, wrap_fn)),
_phantom: PhantomData,
}
}
}
impl<SF1, SF2, F, Fut, Req, In, Res, Err> Clone
for AndThenApplyFnFactory<SF1, SF2, F, Fut, Req, In, Res, Err>
{
fn clone(&self) -> Self {
Self {
srv: self.srv.clone(),
_phantom: PhantomData,
}
}
}
impl<SF1, SF2, F, Fut, Req, In, Res, Err> ServiceFactory<Req>
for AndThenApplyFnFactory<SF1, SF2, F, Fut, Req, In, Res, Err>
where
SF1: ServiceFactory<Req>,
SF1::Config: Clone,
SF2: ServiceFactory<In, Config = SF1::Config, InitError = SF1::InitError>,
F: FnMut(SF1::Response, &mut SF2::Service) -> Fut + Clone,
Fut: Future<Output = Result<Res, Err>>,
Err: From<SF1::Error> + From<SF2::Error>,
{
type Response = Res;
type Error = Err;
type Service = AndThenApplyFn<SF1::Service, SF2::Service, F, Fut, Req, In, Res, Err>;
type Config = SF1::Config;
type InitError = SF1::InitError;
type Future = AndThenApplyFnFactoryResponse<SF1, SF2, F, Fut, Req, In, Res, Err>;
fn new_service(&self, cfg: SF1::Config) -> Self::Future {
let srv = &*self.srv;
AndThenApplyFnFactoryResponse {
s1: None,
s2: None,
wrap_fn: srv.2.clone(),
fut_s1: srv.0.new_service(cfg.clone()),
fut_s2: srv.1.new_service(cfg),
_phantom: PhantomData,
}
}
}
#[pin_project::pin_project]
pub(crate) struct AndThenApplyFnFactoryResponse<SF1, SF2, F, Fut, Req, In, Res, Err>
where
SF1: ServiceFactory<Req>,
SF2: ServiceFactory<In, Config = SF1::Config, InitError = SF1::InitError>,
F: FnMut(SF1::Response, &mut SF2::Service) -> Fut + Clone,
Fut: Future<Output = Result<Res, Err>>,
Err: From<SF1::Error>,
Err: From<SF2::Error>,
{
#[pin]
fut_s1: SF1::Future,
#[pin]
fut_s2: SF2::Future,
wrap_fn: F,
s1: Option<SF1::Service>,
s2: Option<SF2::Service>,
_phantom: PhantomData<In>,
}
impl<SF1, SF2, F, Fut, Req, In, Res, Err> Future
for AndThenApplyFnFactoryResponse<SF1, SF2, F, Fut, Req, In, Res, Err>
where
SF1: ServiceFactory<Req>,
SF2: ServiceFactory<In, Config = SF1::Config, InitError = SF1::InitError>,
F: FnMut(SF1::Response, &mut SF2::Service) -> Fut + Clone,
Fut: Future<Output = Result<Res, Err>>,
Err: From<SF1::Error> + From<SF2::Error>,
{
type Output = Result<
AndThenApplyFn<SF1::Service, SF2::Service, F, Fut, Req, In, Res, Err>,
SF1::InitError,
>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
if this.s1.is_none() {
if let Poll::Ready(service) = this.fut_s1.poll(cx)? {
*this.s1 = Some(service);
}
}
if this.s2.is_none() {
if let Poll::Ready(service) = this.fut_s2.poll(cx)? {
*this.s2 = Some(service);
}
}
if this.s1.is_some() && this.s2.is_some() {
Poll::Ready(Ok(AndThenApplyFn {
svc: Rc::new(RefCell::new((
Option::take(this.s1).unwrap(),
Option::take(this.s2).unwrap(),
this.wrap_fn.clone(),
))),
_phantom: PhantomData,
}))
} else {
Poll::Pending
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use futures_util::future::{lazy, ok, Ready, TryFutureExt};
use crate::{fn_service, pipeline, pipeline_factory, Service, ServiceFactory};
#[derive(Clone)]
struct Srv;
impl Service<u8> for Srv {
type Response = ();
type Error = ();
type Future = Ready<Result<Self::Response, Self::Error>>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, req: u8) -> Self::Future {
let _ = req;
ok(())
}
}
#[actix_rt::test]
async fn test_service() {
let mut srv = pipeline(ok).and_then_apply_fn(Srv, |req: &'static str, s| {
s.call(1).map_ok(move |res| (req, res))
});
let res = lazy(|cx| srv.poll_ready(cx)).await;
assert!(res.is_ready());
let res = srv.call("srv").await;
assert!(res.is_ok());
assert_eq!(res.unwrap(), ("srv", ()));
}
#[actix_rt::test]
async fn test_service_factory() {
let new_srv = pipeline_factory(|| ok::<_, ()>(fn_service(ok))).and_then_apply_fn(
|| ok(Srv),
|req: &'static str, s| s.call(1).map_ok(move |res| (req, res)),
);
let mut srv = new_srv.new_service(()).await.unwrap();
let res = lazy(|cx| srv.poll_ready(cx)).await;
assert!(res.is_ready());
let res = srv.call("srv").await;
assert!(res.is_ok());
assert_eq!(res.unwrap(), ("srv", ()));
}
}

44
actix-service/src/apply.rs
View File

@ -1,11 +1,12 @@
use std::{
use core::{
future::Future,
marker::PhantomData,
pin::Pin,
task::{Context, Poll},
};
use futures_util::ready;
use futures_core::ready;
use pin_project_lite::pin_project;
use super::{IntoService, IntoServiceFactory, Service, ServiceFactory};
@ -94,9 +95,7 @@ where
type Error = Err;
type Future = Fut;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(ready!(self.service.poll_ready(cx)))
}
crate::forward_ready!(service);
fn call(&mut self, req: Req) -> Self::Future {
(self.wrap_fn)(req, &mut self.service)
@ -162,17 +161,18 @@ where
}
}
#[pin_project::pin_project]
pub struct ApplyServiceFactoryResponse<SF, F, Fut, Req, In, Res, Err>
where
SF: ServiceFactory<In, Error = Err>,
F: FnMut(Req, &mut SF::Service) -> Fut,
Fut: Future<Output = Result<Res, Err>>,
{
#[pin]
fut: SF::Future,
wrap_fn: Option<F>,
_phantom: PhantomData<(Req, Res)>,
pin_project! {
pub struct ApplyServiceFactoryResponse<SF, F, Fut, Req, In, Res, Err>
where
SF: ServiceFactory<In, Error = Err>,
F: FnMut(Req, &mut SF::Service) -> Fut,
Fut: Future<Output = Result<Res, Err>>,
{
#[pin]
fut: SF::Future,
wrap_fn: Option<F>,
_phantom: PhantomData<(Req, Res)>,
}
}
impl<SF, F, Fut, Req, In, Res, Err> ApplyServiceFactoryResponse<SF, F, Fut, Req, In, Res, Err>
@ -203,18 +203,18 @@ where
let this = self.project();
let svc = ready!(this.fut.poll(cx))?;
Poll::Ready(Ok(Apply::new(svc, Option::take(this.wrap_fn).unwrap())))
Poll::Ready(Ok(Apply::new(svc, this.wrap_fn.take().unwrap())))
}
}
#[cfg(test)]
mod tests {
use std::task::{Context, Poll};
use core::task::Poll;
use futures_util::future::{lazy, ok, Ready};
use futures_util::future::lazy;
use super::*;
use crate::{pipeline, pipeline_factory, Service, ServiceFactory};
use crate::{ok, pipeline, pipeline_factory, Ready, Service, ServiceFactory};
#[derive(Clone)]
struct Srv;
@ -224,9 +224,7 @@ mod tests {
type Error = ();
type Future = Ready<Result<(), ()>>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
crate::always_ready!();
fn call(&mut self, _: ()) -> Self::Future {
ok(())

85
actix-service/src/apply_cfg.rs
View File

@ -1,9 +1,13 @@
use std::cell::RefCell;
use std::future::Future;
use std::marker::PhantomData;
use std::pin::Pin;
use std::rc::Rc;
use std::task::{Context, Poll};
use alloc::rc::Rc;
use core::{
cell::RefCell,
future::Future,
marker::PhantomData,
pin::Pin,
task::{Context, Poll},
};
use pin_project_lite::pin_project;
use crate::{Service, ServiceFactory};
@ -156,37 +160,42 @@ where
ApplyConfigServiceFactoryResponse {
cfg: Some(cfg),
store: self.srv.clone(),
state: State::A(self.srv.borrow().0.new_service(())),
state: State::A {
fut: self.srv.borrow().0.new_service(()),
},
}
}
}
#[pin_project::pin_project]
struct ApplyConfigServiceFactoryResponse<SF, Req, F, Cfg, Fut, S>
where
SF: ServiceFactory<Req, Config = ()>,
SF::InitError: From<SF::Error>,
F: FnMut(Cfg, &mut SF::Service) -> Fut,
Fut: Future<Output = Result<S, SF::InitError>>,
S: Service<Req>,
{
cfg: Option<Cfg>,
store: Rc<RefCell<(SF, F)>>,
#[pin]
state: State<SF, Fut, S, Req>,
pin_project! {
struct ApplyConfigServiceFactoryResponse<SF, Req, F, Cfg, Fut, S>
where
SF: ServiceFactory<Req, Config = ()>,
SF::InitError: From<SF::Error>,
F: FnMut(Cfg, &mut SF::Service) -> Fut,
Fut: Future<Output = Result<S, SF::InitError>>,
S: Service<Req>,
{
cfg: Option<Cfg>,
store: Rc<RefCell<(SF, F)>>,
#[pin]
state: State<SF, Fut, S, Req>,
}
}
#[pin_project::pin_project(project = StateProj)]
enum State<SF, Fut, S, Req>
where
SF: ServiceFactory<Req, Config = ()>,
SF::InitError: From<SF::Error>,
Fut: Future<Output = Result<S, SF::InitError>>,
S: Service<Req>,
{
A(#[pin] SF::Future),
B(SF::Service),
C(#[pin] Fut),
pin_project! {
#[project = StateProj]
enum State<SF, Fut, S, Req>
where
SF: ServiceFactory<Req, Config = ()>,
SF::InitError: From<SF::Error>,
Fut: Future<Output = Result<S, SF::InitError>>,
S: Service<Req>,
{
A { #[pin] fut: SF::Future },
B { svc: SF::Service },
C { #[pin] fut: Fut },
}
}
impl<SF, Req, F, Cfg, Fut, S> Future
@ -204,25 +213,25 @@ where
let mut this = self.as_mut().project();
match this.state.as_mut().project() {
StateProj::A(fut) => match fut.poll(cx)? {
StateProj::A { fut } => match fut.poll(cx)? {
Poll::Pending => Poll::Pending,
Poll::Ready(srv) => {
this.state.set(State::B(srv));
Poll::Ready(svc) => {
this.state.set(State::B { svc });
self.poll(cx)
}
},
StateProj::B(srv) => match srv.poll_ready(cx)? {
StateProj::B { svc } => match svc.poll_ready(cx)? {
Poll::Ready(_) => {
{
let (_, f) = &mut *this.store.borrow_mut();
let fut = f(this.cfg.take().unwrap(), srv);
this.state.set(State::C(fut));
let fut = f(this.cfg.take().unwrap(), svc);
this.state.set(State::C { fut });
}
self.poll(cx)
}
Poll::Pending => Poll::Pending,
},
StateProj::C(fut) => fut.poll(cx),
StateProj::C { fut } => fut.poll(cx),
}
}
}

31
actix-service/src/boxed.rs
View File

@ -1,8 +1,10 @@
use std::pin::Pin;
use std::task::{Context, Poll};
use std::{future::Future, marker::PhantomData};
use futures_util::future::FutureExt;
use alloc::boxed::Box;
use core::{
future::Future,
marker::PhantomData,
pin::Pin,
task::{Context, Poll},
};
use crate::{Service, ServiceFactory};
@ -28,7 +30,7 @@ where
{
BoxServiceFactory(Box::new(FactoryWrapper {
factory,
_t: std::marker::PhantomData,
_t: PhantomData,
}))
}
@ -75,12 +77,9 @@ where
}
}
struct FactoryWrapper<SF, Req, C>
where
SF: ServiceFactory<Req>,
{
struct FactoryWrapper<SF, Req, Cfg> {
factory: SF,
_t: PhantomData<(C, Req)>,
_t: PhantomData<(Req, Cfg)>,
}
impl<SF, Req, Cfg, Res, Err, InitErr> ServiceFactory<Req> for FactoryWrapper<SF, Req, Cfg>
@ -102,11 +101,11 @@ where
type Future = BoxFuture<Result<Self::Service, Self::InitError>>;
fn new_service(&self, cfg: Cfg) -> Self::Future {
Box::pin(
self.factory
.new_service(cfg)
.map(|res| res.map(ServiceWrapper::boxed)),
)
let fut = self.factory.new_service(cfg);
Box::pin(async {
let res = fut.await;
res.map(ServiceWrapper::boxed)
})
}
}

70
actix-service/src/ext.rs Normal file
View File

@ -0,0 +1,70 @@
use crate::{dev, Service, ServiceFactory};
pub trait ServiceExt<Req>: Service<Req> {
/// Map this service's output to a different type, returning a new service
/// of the resulting type.
///
/// This function is similar to the `Option::map` or `Iterator::map` where
/// it will change the type of the underlying service.
///
/// Note that this function consumes the receiving service and returns a
/// wrapped version of it, similar to the existing `map` methods in the
/// standard library.
fn map<F, R>(self, f: F) -> dev::Map<Self, F, Req, R>
where
Self: Sized,
F: FnMut(Self::Response) -> R,
{
dev::Map::new(self, f)
}
/// Map this service's error to a different error, returning a new service.
///
/// This function is similar to the `Result::map_err` where it will change
/// the error type of the underlying service. For example, this can be useful to
/// ensure that services have the same error type.
///
/// Note that this function consumes the receiving service and returns a
/// wrapped version of it.
fn map_err<F, E>(self, f: F) -> dev::MapErr<Self, Req, F, E>
where
Self: Sized,
F: Fn(Self::Error) -> E,
{
dev::MapErr::new(self, f)
}
}
impl<S, Req> ServiceExt<Req> for S where S: Service<Req> {}
pub trait ServiceFactoryExt<Req>: ServiceFactory<Req> {
/// Map this service's output to a different type, returning a new service
/// of the resulting type.
fn map<F, R>(self, f: F) -> crate::map::MapServiceFactory<Self, F, Req, R>
where
Self: Sized,
F: FnMut(Self::Response) -> R + Clone,
{
crate::map::MapServiceFactory::new(self, f)
}
/// Map this service's error to a different error, returning a new service.
fn map_err<F, E>(self, f: F) -> crate::map_err::MapErrServiceFactory<Self, Req, F, E>
where
Self: Sized,
F: Fn(Self::Error) -> E + Clone,
{
crate::map_err::MapErrServiceFactory::new(self, f)
}
/// Map this factory's init error to a different error, returning a new service.
fn map_init_err<F, E>(self, f: F) -> crate::map_init_err::MapInitErr<Self, F, Req, E>
where
Self: Sized,
F: Fn(Self::InitError) -> E + Clone,
{
crate::map_init_err::MapInitErr::new(self, f)
}
}
impl<S, Req> ServiceFactoryExt<Req> for S where S: ServiceFactory<Req> {}

22
actix-service/src/fn_service.rs
View File

@ -1,10 +1,6 @@
use std::future::Future;
use std::marker::PhantomData;
use std::task::{Context, Poll};
use core::{future::Future, marker::PhantomData, task::Poll};
use futures_util::future::{ok, Ready};
use crate::{IntoService, IntoServiceFactory, Service, ServiceFactory};
use crate::{ok, IntoService, IntoServiceFactory, Ready, Service, ServiceFactory};
/// Create `ServiceFactory` for function that can act as a `Service`
pub fn fn_service<F, Fut, Req, Res, Err, Cfg>(
@ -143,9 +139,7 @@ where
type Error = Err;
type Future = Fut;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
crate::always_ready!();
fn call(&mut self, req: Req) -> Self::Future {
(self.f)(req)
@ -200,9 +194,7 @@ where
type Error = Err;
type Future = Fut;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
crate::always_ready!();
fn call(&mut self, req: Req) -> Self::Future {
(self.f)(req)
@ -361,12 +353,12 @@ where
#[cfg(test)]
mod tests {
use std::task::Poll;
use core::task::Poll;
use futures_util::future::{lazy, ok};
use futures_util::future::lazy;
use super::*;
use crate::{Service, ServiceFactory};
use crate::{ok, Service, ServiceFactory};
#[actix_rt::test]
async fn test_fn_service() {

106
actix-service/src/lib.rs
View File

@ -1,38 +1,47 @@
//! See [`Service`] docs for information on this crate's foundational trait.
#![no_std]
#![deny(rust_2018_idioms, nonstandard_style)]
#![allow(clippy::type_complexity)]
#![doc(html_logo_url = "https://actix.rs/img/logo.png")]
#![doc(html_favicon_url = "https://actix.rs/favicon.ico")]
use std::cell::RefCell;
use std::future::Future;
use std::rc::Rc;
use std::sync::Arc;
use std::task::{self, Context, Poll};
extern crate alloc;
use alloc::{boxed::Box, rc::Rc, sync::Arc};
use core::{
cell::RefCell,
future::Future,
task::{self, Context, Poll},
};
mod and_then;
mod and_then_apply_fn;
mod apply;
mod apply_cfg;
pub mod boxed;
mod ext;
mod fn_service;
mod map;
mod map_config;
mod map_err;
mod map_init_err;
mod pipeline;
mod ready;
mod then;
mod transform;
mod transform_err;
pub use self::apply::{apply_fn, apply_fn_factory};
pub use self::apply_cfg::{apply_cfg, apply_cfg_factory};
pub use self::ext::{ServiceExt, ServiceFactoryExt};
pub use self::fn_service::{fn_factory, fn_factory_with_config, fn_service};
pub use self::map_config::{map_config, unit_config};
pub use self::pipeline::{pipeline, pipeline_factory, Pipeline, PipelineFactory};
pub use self::transform::{apply, Transform};
#[allow(unused_imports)]
use self::ready::{err, ok, ready, Ready};
/// An asynchronous operation from `Request` to a `Response`.
///
/// The `Service` trait models a request/response interaction, receiving requests and returning
@ -107,39 +116,6 @@ pub trait Service<Req> {
/// Calling `call` without calling `poll_ready` is permitted. The
/// implementation must be resilient to this fact.
fn call(&mut self, req: Req) -> Self::Future;
/// Map this service's output to a different type, returning a new service
/// of the resulting type.
///
/// This function is similar to the `Option::map` or `Iterator::map` where
/// it will change the type of the underlying service.
///
/// Note that this function consumes the receiving service and returns a
/// wrapped version of it, similar to the existing `map` methods in the
/// standard library.
fn map<F, R>(self, f: F) -> crate::dev::Map<Self, F, Req, R>
where
Self: Sized,
F: FnMut(Self::Response) -> R,
{
crate::dev::Map::new(self, f)
}
/// Map this service's error to a different error, returning a new service.
///
/// This function is similar to the `Result::map_err` where it will change
/// the error type of the underlying service. For example, this can be useful to
/// ensure that services have the same error type.
///
/// Note that this function consumes the receiving service and returns a
/// wrapped version of it.
fn map_err<F, E>(self, f: F) -> crate::dev::MapErr<Self, Req, F, E>
where
Self: Sized,
F: Fn(Self::Error) -> E,
{
crate::dev::MapErr::new(self, f)
}
}
/// Factory for creating `Service`s.
@ -172,34 +148,6 @@ pub trait ServiceFactory<Req> {
/// Create and return a new service asynchronously.
fn new_service(&self, cfg: Self::Config) -> Self::Future;
/// Map this service's output to a different type, returning a new service
/// of the resulting type.
fn map<F, R>(self, f: F) -> crate::map::MapServiceFactory<Self, F, Req, R>
where
Self: Sized,
F: FnMut(Self::Response) -> R + Clone,
{
crate::map::MapServiceFactory::new(self, f)
}
/// Map this service's error to a different error, returning a new service.
fn map_err<F, E>(self, f: F) -> crate::map_err::MapErrServiceFactory<Self, Req, F, E>
where
Self: Sized,
F: Fn(Self::Error) -> E + Clone,
{
crate::map_err::MapErrServiceFactory::new(self, f)
}
/// Map this factory's init error to a different error, returning a new service.
fn map_init_err<F, E>(self, f: F) -> crate::map_init_err::MapInitErr<Self, F, Req, E>
where
Self: Sized,
F: Fn(Self::InitError) -> E + Clone,
{
crate::map_init_err::MapInitErr::new(self, f)
}
}
impl<'a, S, Req> Service<Req> for &'a mut S
@ -359,3 +307,27 @@ pub mod dev {
pub use crate::transform::ApplyTransform;
pub use crate::transform_err::TransformMapInitErr;
}
#[macro_export]
macro_rules! always_ready {
() => {
fn poll_ready(
&mut self,
_: &mut ::core::task::Context<'_>,
) -> ::core::task::Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
};
}
#[macro_export]
macro_rules! forward_ready {
($field:ident) => {
fn poll_ready(
&mut self,
cx: &mut ::core::task::Context<'_>,
) -> ::core::task::Poll<Result<(), Self::Error>> {
self.$field.poll_ready(cx)
}
};
}

64
actix-service/src/map.rs
View File

@ -1,7 +1,11 @@
use std::future::Future;
use std::marker::PhantomData;
use std::pin::Pin;
use std::task::{Context, Poll};
use core::{
future::Future,
marker::PhantomData,
pin::Pin,
task::{Context, Poll},
};
use pin_project_lite::pin_project;
use super::{Service, ServiceFactory};
@ -52,24 +56,23 @@ where
type Error = A::Error;
type Future = MapFuture<A, F, Req, Res>;
fn poll_ready(&mut self, ctx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.service.poll_ready(ctx)
}
crate::forward_ready!(service);
fn call(&mut self, req: Req) -> Self::Future {
MapFuture::new(self.service.call(req), self.f.clone())
}
}
#[pin_project::pin_project]
pub struct MapFuture<A, F, Req, Res>
where
A: Service<Req>,
F: FnMut(A::Response) -> Res,
{
f: F,
#[pin]
fut: A::Future,
pin_project! {
pub struct MapFuture<A, F, Req, Res>
where
A: Service<Req>,
F: FnMut(A::Response) -> Res,
{
f: F,
#[pin]
fut: A::Future,
}
}
impl<A, F, Req, Res> MapFuture<A, F, Req, Res>
@ -154,15 +157,16 @@ where
}
}
#[pin_project::pin_project]
pub struct MapServiceFuture<A, F, Req, Res>
where
A: ServiceFactory<Req>,
F: FnMut(A::Response) -> Res,
{
#[pin]
fut: A::Future,
f: Option<F>,
pin_project! {
pub struct MapServiceFuture<A, F, Req, Res>
where
A: ServiceFactory<Req>,
F: FnMut(A::Response) -> Res,
{
#[pin]
fut: A::Future,
f: Option<F>,
}
}
impl<A, F, Req, Res> MapServiceFuture<A, F, Req, Res>
@ -195,10 +199,12 @@ where
#[cfg(test)]
mod tests {
use futures_util::future::{lazy, ok, Ready};
use futures_util::future::lazy;
use super::*;
use crate::{IntoServiceFactory, Service, ServiceFactory};
use crate::{
ok, IntoServiceFactory, Ready, Service, ServiceExt, ServiceFactory, ServiceFactoryExt,
};
struct Srv;
@ -207,9 +213,7 @@ mod tests {
type Error = ();
type Future = Ready<Result<(), ()>>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
crate::always_ready!();
fn call(&mut self, _: ()) -> Self::Future {
ok(())

2
actix-service/src/map_config.rs
View File

@ -1,4 +1,4 @@
use std::marker::PhantomData;
use core::marker::PhantomData;
use super::{IntoServiceFactory, ServiceFactory};

57
actix-service/src/map_err.rs
View File

@ -1,7 +1,11 @@
use std::future::Future;
use std::marker::PhantomData;
use std::pin::Pin;
use std::task::{Context, Poll};
use core::{
future::Future,
marker::PhantomData,
pin::Pin,
task::{Context, Poll},
};
use pin_project_lite::pin_project;
use super::{Service, ServiceFactory};
@ -62,15 +66,16 @@ where
}
}
#[pin_project::pin_project]
pub struct MapErrFuture<A, Req, F, E>
where
A: Service<Req>,
F: Fn(A::Error) -> E,
{
f: F,
#[pin]
fut: A::Future,
pin_project! {
pub struct MapErrFuture<A, Req, F, E>
where
A: Service<Req>,
F: Fn(A::Error) -> E,
{
f: F,
#[pin]
fut: A::Future,
}
}
impl<A, Req, F, E> MapErrFuture<A, Req, F, E>
@ -157,15 +162,16 @@ where
}
}
#[pin_project::pin_project]
pub struct MapErrServiceFuture<A, Req, F, E>
where
A: ServiceFactory<Req>,
F: Fn(A::Error) -> E,
{
#[pin]
fut: A::Future,
f: F,
pin_project! {
pub struct MapErrServiceFuture<A, Req, F, E>
where
A: ServiceFactory<Req>,
F: Fn(A::Error) -> E,
{
#[pin]
fut: A::Future,
f: F,
}
}
impl<A, Req, F, E> MapErrServiceFuture<A, Req, F, E>
@ -197,10 +203,13 @@ where
#[cfg(test)]
mod tests {
use futures_util::future::{err, lazy, ok, Ready};
use futures_util::future::lazy;
use super::*;
use crate::{IntoServiceFactory, Service, ServiceFactory};
use crate::{
err, ok, IntoServiceFactory, Ready, Service, ServiceExt, ServiceFactory,
ServiceFactoryExt,
};
struct Srv;

31
actix-service/src/map_init_err.rs
View File

@ -1,7 +1,11 @@
use std::future::Future;
use std::marker::PhantomData;
use std::pin::Pin;
use std::task::{Context, Poll};
use core::{
future::Future,
marker::PhantomData,
pin::Pin,
task::{Context, Poll},
};
use pin_project_lite::pin_project;
use super::ServiceFactory;
@ -59,15 +63,16 @@ where
}
}
#[pin_project::pin_project]
pub struct MapInitErrFuture<A, F, Req, E>
where
A: ServiceFactory<Req>,
F: Fn(A::InitError) -> E,
{
f: F,
#[pin]
fut: A::Future,
pin_project! {
pub struct MapInitErrFuture<A, F, Req, E>
where
A: ServiceFactory<Req>,
F: Fn(A::InitError) -> E,
{
f: F,
#[pin]
fut: A::Future,
}
}
impl<A, F, Req, E> MapInitErrFuture<A, F, Req, E>

62
actix-service/src/pipeline.rs
View File

@ -1,8 +1,9 @@
use std::task::{Context, Poll};
use std::{future::Future, marker::PhantomData};
use core::{
marker::PhantomData,
task::{Context, Poll},
};
use crate::and_then::{AndThenService, AndThenServiceFactory};
use crate::and_then_apply_fn::{AndThenApplyFn, AndThenApplyFnFactory};
use crate::map::{Map, MapServiceFactory};
use crate::map_err::{MapErr, MapErrServiceFactory};
use crate::map_init_err::MapInitErr;
@ -67,28 +68,6 @@ where
}
}
/// Apply function to specified service and use it as a next service in chain.
///
/// Short version of `pipeline_factory(...).and_then(apply_fn(...))`
pub fn and_then_apply_fn<I, S1, F, Fut, In, Res, Err>(
self,
service: I,
wrap_fn: F,
) -> Pipeline<impl Service<Req, Response = Res, Error = Err> + Clone, Req>
where
Self: Sized,
I: IntoService<S1, In>,
S1: Service<In>,
F: FnMut(S::Response, &mut S1) -> Fut,
Fut: Future<Output = Result<Res, Err>>,
Err: From<S::Error> + From<S1::Error>,
{
Pipeline {
service: AndThenApplyFn::new(self.service, service.into_service(), wrap_fn),
_phantom: PhantomData,
}
}
/// Chain on a computation for when a call to the service finished,
/// passing the result of the call to the next service `U`.
///
@ -219,39 +198,6 @@ where
}
}
/// Apply function to specified service and use it as a next service in chain.
///
/// Short version of `pipeline_factory(...).and_then(apply_fn_factory(...))`
pub fn and_then_apply_fn<I, SF1, Fut, F, In, Res, Err>(
self,
factory: I,
wrap_fn: F,
) -> PipelineFactory<
impl ServiceFactory<
Req,
Response = Res,
Error = Err,
Config = SF::Config,
InitError = SF::InitError,
Service = impl Service<Req, Response = Res, Error = Err> + Clone,
> + Clone,
Req,
>
where
Self: Sized,
SF::Config: Clone,
I: IntoServiceFactory<SF1, In>,
SF1: ServiceFactory<In, Config = SF::Config, InitError = SF::InitError>,
F: FnMut(SF::Response, &mut SF1::Service) -> Fut + Clone,
Fut: Future<Output = Result<Res, Err>>,
Err: From<SF::Error> + From<SF1::Error>,
{
PipelineFactory {
factory: AndThenApplyFnFactory::new(self.factory, factory.into_factory(), wrap_fn),
_phantom: PhantomData,
}
}
/// Create `NewService` to chain on a computation for when a call to the
/// service finished, passing the result of the call to the next
/// service `U`.

54
actix-service/src/ready.rs Normal file
View File

@ -0,0 +1,54 @@
//! When MSRV is 1.48, replace with `core::future::Ready` and `core::future::ready()`.
use core::{
future::Future,
pin::Pin,
task::{Context, Poll},
};
/// Future for the [`ready`](ready()) function.
#[derive(Debug, Clone)]
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct Ready<T> {
val: Option<T>,
}
impl<T> Ready<T> {
/// Unwraps the value from this immediately ready future.
#[inline]
pub fn into_inner(mut self) -> T {
self.val.take().unwrap()
}
}
impl<T> Unpin for Ready<T> {}
impl<T> Future for Ready<T> {
type Output = T;
#[inline]
fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<T> {
let val = self.val.take().expect("Ready can not be polled twice.");
Poll::Ready(val)
}
}
/// Creates a future that is immediately ready with a value.
#[allow(dead_code)]
pub(crate) fn ready<T>(val: T) -> Ready<T> {
Ready { val: Some(val) }
}
/// Create a future that is immediately ready with a success value.
#[allow(dead_code)]
pub(crate) fn ok<T, E>(val: T) -> Ready<Result<T, E>> {
Ready { val: Some(Ok(val)) }
}
/// Create a future that is immediately ready with an error value.
#[allow(dead_code)]
pub(crate) fn err<T, E>(err: E) -> Ready<Result<T, E>> {
Ready {
val: Some(Err(err)),
}
}

110
actix-service/src/then.rs
View File

@ -1,8 +1,13 @@
use std::future::Future;
use std::pin::Pin;
use std::rc::Rc;
use std::task::{Context, Poll};
use std::{cell::RefCell, marker::PhantomData};
use alloc::rc::Rc;
use core::{
cell::RefCell,
future::Future,
marker::PhantomData,
pin::Pin,
task::{Context, Poll},
};
use pin_project_lite::pin_project;
use super::{Service, ServiceFactory};
@ -50,30 +55,36 @@ where
fn call(&mut self, req: Req) -> Self::Future {
ThenServiceResponse {
state: State::A(self.0.borrow_mut().0.call(req), Some(self.0.clone())),
state: State::A {
fut: self.0.borrow_mut().0.call(req),
b: Some(self.0.clone()),
},
}
}
}
#[pin_project::pin_project]
pub(crate) struct ThenServiceResponse<A, B, Req>
where
A: Service<Req>,
B: Service<Result<A::Response, A::Error>>,
{
#[pin]
state: State<A, B, Req>,
pin_project! {
pub(crate) struct ThenServiceResponse<A, B, Req>
where
A: Service<Req>,
B: Service<Result<A::Response, A::Error>>,
{
#[pin]
state: State<A, B, Req>,
}
}
#[pin_project::pin_project(project = StateProj)]
enum State<A, B, Req>
where
A: Service<Req>,
B: Service<Result<A::Response, A::Error>>,
{
A(#[pin] A::Future, Option<Rc<RefCell<(A, B)>>>),
B(#[pin] B::Future),
Empty,
pin_project! {
#[project = StateProj]
enum State<A, B, Req>
where
A: Service<Req>,
B: Service<Result<A::Response, A::Error>>,
{
A { #[pin] fut: A::Future, b: Option<Rc<RefCell<(A, B)>>> },
B { #[pin] fut: B::Future },
Empty,
}
}
impl<A, B, Req> Future for ThenServiceResponse<A, B, Req>
@ -87,17 +98,17 @@ where
let mut this = self.as_mut().project();
match this.state.as_mut().project() {
StateProj::A(fut, b) => match fut.poll(cx) {
StateProj::A { fut, b } => match fut.poll(cx) {
Poll::Ready(res) => {
let b = b.take().unwrap();
this.state.set(State::Empty); // drop fut A
let fut = b.borrow_mut().1.call(res);
this.state.set(State::B(fut));
this.state.set(State::B { fut });
self.poll(cx)
}
Poll::Pending => Poll::Pending,
},
StateProj::B(fut) => fut.poll(cx).map(|r| {
StateProj::B { fut } => fut.poll(cx).map(|r| {
this.state.set(State::Empty);
r
}),
@ -159,23 +170,24 @@ impl<A, B, Req> Clone for ThenServiceFactory<A, B, Req> {
}
}
#[pin_project::pin_project]
pub(crate) struct ThenServiceFactoryResponse<A, B, Req>
where
A: ServiceFactory<Req>,
B: ServiceFactory<
Result<A::Response, A::Error>,
Config = A::Config,
Error = A::Error,
InitError = A::InitError,
>,
{
#[pin]
fut_b: B::Future,
#[pin]
fut_a: A::Future,
a: Option<A::Service>,
b: Option<B::Service>,
pin_project! {
pub(crate) struct ThenServiceFactoryResponse<A, B, Req>
where
A: ServiceFactory<Req>,
B: ServiceFactory<
Result<A::Response, A::Error>,
Config = A::Config,
Error = A::Error,
InitError = A::InitError,
>,
{
#[pin]
fut_b: B::Future,
#[pin]
fut_a: A::Future,
a: Option<A::Service>,
b: Option<B::Service>,
}
}
impl<A, B, Req> ThenServiceFactoryResponse<A, B, Req>
@ -236,13 +248,15 @@ where
#[cfg(test)]
mod tests {
use std::cell::Cell;
use std::rc::Rc;
use std::task::{Context, Poll};
use alloc::rc::Rc;
use core::{
cell::Cell,
task::{Context, Poll},
};
use futures_util::future::{err, lazy, ok, ready, Ready};
use futures_util::future::lazy;
use crate::{pipeline, pipeline_factory, Service, ServiceFactory};
use crate::{err, ok, pipeline, pipeline_factory, ready, Ready, Service, ServiceFactory};
#[derive(Clone)]
struct Srv1(Rc<Cell<usize>>);

61
actix-service/src/transform.rs
View File

@ -1,8 +1,12 @@
use std::pin::Pin;
use std::rc::Rc;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::{future::Future, marker::PhantomData};
use alloc::{rc::Rc, sync::Arc};
use core::{
future::Future,
marker::PhantomData,
pin::Pin,
task::{Context, Poll},
};
use pin_project_lite::pin_project;
use crate::transform_err::TransformMapInitErr;
use crate::{IntoServiceFactory, Service, ServiceFactory};
@ -185,30 +189,35 @@ where
fn new_service(&self, cfg: S::Config) -> Self::Future {
ApplyTransformFuture {
store: self.0.clone(),
state: ApplyTransformFutureState::A(self.0.as_ref().1.new_service(cfg)),
state: ApplyTransformFutureState::A {
fut: self.0.as_ref().1.new_service(cfg),
},
}
}
}
#[pin_project::pin_project]
pub struct ApplyTransformFuture<T, S, Req>
where
S: ServiceFactory<Req>,
T: Transform<S::Service, Req, InitError = S::InitError>,
{
store: Rc<(T, S)>,
#[pin]
state: ApplyTransformFutureState<T, S, Req>,
pin_project! {
pub struct ApplyTransformFuture<T, S, Req>
where
S: ServiceFactory<Req>,
T: Transform<S::Service, Req, InitError = S::InitError>,
{
store: Rc<(T, S)>,
#[pin]
state: ApplyTransformFutureState<T, S, Req>,
}
}
#[pin_project::pin_project(project = ApplyTransformFutureStateProj)]
pub enum ApplyTransformFutureState<T, S, Req>
where
S: ServiceFactory<Req>,
T: Transform<S::Service, Req, InitError = S::InitError>,
{
A(#[pin] S::Future),
B(#[pin] T::Future),
pin_project! {
#[project = ApplyTransformFutureStateProj]
pub enum ApplyTransformFutureState<T, S, Req>
where
S: ServiceFactory<Req>,
T: Transform<S::Service, Req, InitError = S::InitError>,
{
A { #[pin] fut: S::Future },
B { #[pin] fut: T::Future },
}
}
impl<T, S, Req> Future for ApplyTransformFuture<T, S, Req>
@ -222,15 +231,15 @@ where
let mut this = self.as_mut().project();
match this.state.as_mut().project() {
ApplyTransformFutureStateProj::A(fut) => match fut.poll(cx)? {
ApplyTransformFutureStateProj::A { fut } => match fut.poll(cx)? {
Poll::Ready(srv) => {
let fut = this.store.0.new_transform(srv);
this.state.set(ApplyTransformFutureState::B(fut));
this.state.set(ApplyTransformFutureState::B { fut });
self.poll(cx)
}
Poll::Pending => Poll::Pending,
},
ApplyTransformFutureStateProj::B(fut) => fut.poll(cx),
ApplyTransformFutureStateProj::B { fut } => fut.poll(cx),
}
}
}

27
actix-service/src/transform_err.rs
View File

@ -1,7 +1,11 @@
use std::future::Future;
use std::marker::PhantomData;
use std::pin::Pin;
use std::task::{Context, Poll};
use core::{
future::Future,
marker::PhantomData,
pin::Pin,
task::{Context, Poll},
};
use pin_project_lite::pin_project;
use super::Transform;
@ -63,15 +67,16 @@ where
}
}
#[pin_project::pin_project]
pub struct TransformMapInitErrFuture<T, S, F, E, Req>
where
pin_project! {
pub struct TransformMapInitErrFuture<T, S, F, E, Req>
where
T: Transform<S, Req>,
F: Fn(T::InitError) -> E,
{
#[pin]
fut: T::Future,
f: F,
{
#[pin]
fut: T::Future,
f: F,
}
}
impl<T, S, F, E, Req> Future for TransformMapInitErrFuture<T, S, F, E, Req>

2
actix-tls/Cargo.toml
View File

@ -37,7 +37,7 @@ nativetls = ["native-tls", "tokio-native-tls"]
[dependencies]
actix-service = "1.0.0"
actix-codec = "0.3.0"
actix-utils = "3.0.0"
actix-utils = "2.0.0"
futures-util = { version = "0.3.7", default-features = false }

44
actix-tls/src/nativetls.rs
View File

@ -1,5 +1,4 @@
use std::future::Future;
use std::marker::PhantomData;
use std::pin::Pin;
use std::task::{Context, Poll};
@ -16,45 +15,36 @@ use crate::MAX_CONN_COUNTER;
/// Accept TLS connections via `native-tls` package.
///
/// `nativetls` feature enables this `Acceptor` type.
pub struct Acceptor<T> {
pub struct Acceptor {
acceptor: TlsAcceptor,
io: PhantomData<T>,
}
impl<T> Acceptor<T>
where
T: AsyncRead + AsyncWrite + Unpin,
{
impl Acceptor {
/// Create `native-tls` based `Acceptor` service factory.
#[inline]
pub fn new(acceptor: TlsAcceptor) -> Self {
Acceptor {
acceptor,
io: PhantomData,
}
Acceptor { acceptor }
}
}
impl<T> Clone for Acceptor<T> {
impl Clone for Acceptor {
#[inline]
fn clone(&self) -> Self {
Self {
acceptor: self.acceptor.clone(),
io: PhantomData,
}
}
}
impl<T> ServiceFactory for Acceptor<T>
impl<Req> ServiceFactory<Req> for Acceptor
where
T: AsyncRead + AsyncWrite + Unpin + 'static,
Req: AsyncRead + AsyncWrite + Unpin + 'static,
{
type Request = T;
type Response = TlsStream<T>;
type Response = TlsStream<Req>;
type Error = Error;
type Config = ();
type Service = NativeTlsAcceptorService<T>;
type Service = NativeTlsAcceptorService;
type InitError = ();
type Future = Ready<Result<Self::Service, Self::InitError>>;
@ -63,23 +53,20 @@ where
ready(Ok(NativeTlsAcceptorService {
acceptor: self.acceptor.clone(),
conns: conns.clone(),
io: PhantomData,
}))
})
}
}
pub struct NativeTlsAcceptorService<T> {
pub struct NativeTlsAcceptorService {
acceptor: TlsAcceptor,
io: PhantomData<T>,
conns: Counter,
}
impl<T> Clone for NativeTlsAcceptorService<T> {
impl Clone for NativeTlsAcceptorService {
fn clone(&self) -> Self {
Self {
acceptor: self.acceptor.clone(),
io: PhantomData,
conns: self.conns.clone(),
}
}
@ -87,14 +74,13 @@ impl<T> Clone for NativeTlsAcceptorService<T> {
type LocalBoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + 'a>>;
impl<T> Service for NativeTlsAcceptorService<T>
impl<Req> Service<Req> for NativeTlsAcceptorService
where
T: AsyncRead + AsyncWrite + Unpin + 'static,
Req: AsyncRead + AsyncWrite + Unpin + 'static,
{
type Request = T;
type Response = TlsStream<T>;
type Response = TlsStream<Req>;
type Error = Error;
type Future = LocalBoxFuture<'static, Result<TlsStream<T>, Error>>;
type Future = LocalBoxFuture<'static, Result<TlsStream<Req>, Error>>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
if self.conns.available(cx) {
@ -104,7 +90,7 @@ where
}
}
fn call(&mut self, req: Self::Request) -> Self::Future {
fn call(&mut self, req: Req) -> Self::Future {
let guard = self.conns.get();
let this = self.clone();
Box::pin(async move {

52
actix-tls/src/openssl.rs
View File

@ -1,5 +1,4 @@
use std::future::Future;
use std::marker::PhantomData;
use std::pin::Pin;
use std::task::{Context, Poll};
@ -17,38 +16,32 @@ use crate::MAX_CONN_COUNTER;
/// Accept TLS connections via `openssl` package.
///
/// `openssl` feature enables this `Acceptor` type.
pub struct Acceptor<T: AsyncRead + AsyncWrite> {
pub struct Acceptor {
acceptor: SslAcceptor,
io: PhantomData<T>,
}
impl<T: AsyncRead + AsyncWrite> Acceptor<T> {
impl Acceptor {
/// Create OpenSSL based `Acceptor` service factory.
#[inline]
pub fn new(acceptor: SslAcceptor) -> Self {
Acceptor {
acceptor,
io: PhantomData,
}
Acceptor { acceptor }
}
}
impl<T: AsyncRead + AsyncWrite> Clone for Acceptor<T> {
impl Clone for Acceptor {
#[inline]
fn clone(&self) -> Self {
Self {
acceptor: self.acceptor.clone(),
io: PhantomData,
}
}
}
impl<T: AsyncRead + AsyncWrite + Unpin + 'static> ServiceFactory for Acceptor<T> {
type Request = T;
impl<T: AsyncRead + AsyncWrite + Unpin + 'static> ServiceFactory<T> for Acceptor {
type Response = SslStream<T>;
type Error = Error;
type Config = ();
type Service = AcceptorService<T>;
type Service = AcceptorService;
type InitError = ();
type Future = Ready<Result<Self::Service, Self::InitError>>;
@ -57,23 +50,20 @@ impl<T: AsyncRead + AsyncWrite + Unpin + 'static> ServiceFactory for Acceptor<T>
ready(Ok(AcceptorService {
acceptor: self.acceptor.clone(),
conns: conns.clone(),
io: PhantomData,
}))
})
}
}
pub struct AcceptorService<T> {
pub struct AcceptorService {
acceptor: SslAcceptor,
conns: Counter,
io: PhantomData<T>,
}
impl<T: AsyncRead + AsyncWrite + Unpin + 'static> Service for AcceptorService<T> {
type Request = T;
type Response = SslStream<T>;
impl<Req: AsyncRead + AsyncWrite + Unpin + 'static> Service<Req> for AcceptorService {
type Response = SslStream<Req>;
type Error = Error;
type Future = AcceptorServiceResponse<T>;
type Future = AcceptorServiceResponse<Req>;
fn poll_ready(&mut self, ctx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
if self.conns.available(ctx) {
@ -83,7 +73,7 @@ impl<T: AsyncRead + AsyncWrite + Unpin + 'static> Service for AcceptorService<T>
}
}
fn call(&mut self, req: Self::Request) -> Self::Future {
fn call(&mut self, req: Req) -> Self::Future {
match self.ssl_stream(req) {
Ok(stream) => {
let guard = self.conns.get();
@ -94,27 +84,33 @@ impl<T: AsyncRead + AsyncWrite + Unpin + 'static> Service for AcceptorService<T>
}
}
impl<T: AsyncRead + AsyncWrite + Unpin + 'static> AcceptorService<T> {
impl AcceptorService {
// construct a new SslStream.
// At this point the SslStream does not perform any IO.
// The handshake would happen later in AcceptorServiceResponse
fn ssl_stream(&self, stream: T) -> Result<SslStream<T>, Error> {
fn ssl_stream<Req: AsyncRead + AsyncWrite + Unpin + 'static>(
&self,
stream: Req,
) -> Result<SslStream<Req>, Error> {
let ssl = Ssl::new(self.acceptor.context())?;
let stream = SslStream::new(ssl, stream)?;
Ok(stream)
}
}
pub enum AcceptorServiceResponse<T>
pub enum AcceptorServiceResponse<Req>
where
T: AsyncRead + AsyncWrite,
Req: AsyncRead + AsyncWrite,
{
Accept(Option<SslStream<T>>, Option<CounterGuard>),
Accept(Option<SslStream<Req>>, Option<CounterGuard>),
Error(Option<Error>),
}
impl<T: AsyncRead + AsyncWrite + Unpin> Future for AcceptorServiceResponse<T> {
type Output = Result<SslStream<T>, Error>;
impl<Req: AsyncRead + AsyncWrite + Unpin> Future for AcceptorServiceResponse<Req>
where
Req: AsyncRead + AsyncWrite + Unpin,
{
type Output = Result<SslStream<Req>, Error>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
match self.get_mut() {

40
actix-tls/src/rustls.rs
View File

@ -1,6 +1,5 @@
use std::future::Future;
use std::io;
use std::marker::PhantomData;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
@ -20,39 +19,35 @@ use crate::MAX_CONN_COUNTER;
/// Accept TLS connections via `rustls` package.
///
/// `rustls` feature enables this `Acceptor` type.
pub struct Acceptor<T> {
pub struct Acceptor {
config: Arc<ServerConfig>,
io: PhantomData<T>,
}
impl<T: AsyncRead + AsyncWrite> Acceptor<T> {
impl Acceptor {
/// Create Rustls based `Acceptor` service factory.
#[inline]
pub fn new(config: ServerConfig) -> Self {
Acceptor {
config: Arc::new(config),
io: PhantomData,
}
}
}
impl<T> Clone for Acceptor<T> {
impl Clone for Acceptor {
#[inline]
fn clone(&self) -> Self {
Self {
config: self.config.clone(),
io: PhantomData,
}
}
}
impl<T: AsyncRead + AsyncWrite + Unpin> ServiceFactory for Acceptor<T> {
type Request = T;
type Response = TlsStream<T>;
impl<Req: AsyncRead + AsyncWrite + Unpin> ServiceFactory<Req> for Acceptor {
type Response = TlsStream<Req>;
type Error = io::Error;
type Config = ();
type Service = AcceptorService<T>;
type Service = AcceptorService;
type InitError = ();
type Future = Ready<Result<Self::Service, Self::InitError>>;
@ -61,24 +56,21 @@ impl<T: AsyncRead + AsyncWrite + Unpin> ServiceFactory for Acceptor<T> {
ready(Ok(AcceptorService {
acceptor: self.config.clone().into(),
conns: conns.clone(),
io: PhantomData,
}))
})
}
}
/// Rustls based `Acceptor` service
pub struct AcceptorService<T> {
pub struct AcceptorService {
acceptor: TlsAcceptor,
io: PhantomData<T>,
conns: Counter,
}
impl<T: AsyncRead + AsyncWrite + Unpin> Service for AcceptorService<T> {
type Request = T;
type Response = TlsStream<T>;
impl<Req: AsyncRead + AsyncWrite + Unpin> Service<Req> for AcceptorService {
type Response = TlsStream<Req>;
type Error = io::Error;
type Future = AcceptorServiceFut<T>;
type Future = AcceptorServiceFut<Req>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
if self.conns.available(cx) {
@ -88,7 +80,7 @@ impl<T: AsyncRead + AsyncWrite + Unpin> Service for AcceptorService<T> {
}
}
fn call(&mut self, req: Self::Request) -> Self::Future {
fn call(&mut self, req: Req) -> Self::Future {
AcceptorServiceFut {
_guard: self.conns.get(),
fut: self.acceptor.accept(req),
@ -96,16 +88,16 @@ impl<T: AsyncRead + AsyncWrite + Unpin> Service for AcceptorService<T> {
}
}
pub struct AcceptorServiceFut<T>
pub struct AcceptorServiceFut<Req>
where
T: AsyncRead + AsyncWrite + Unpin,
Req: AsyncRead + AsyncWrite + Unpin,
{
fut: Accept<T>,
fut: Accept<Req>,
_guard: CounterGuard,
}
impl<T: AsyncRead + AsyncWrite + Unpin> Future for AcceptorServiceFut<T> {
type Output = Result<TlsStream<T>, io::Error>;
impl<Req: AsyncRead + AsyncWrite + Unpin> Future for AcceptorServiceFut<Req> {
type Output = Result<TlsStream<Req>, io::Error>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.get_mut();

7
actix-tracing/src/lib.rs
View File

@ -4,8 +4,7 @@
#![doc(html_logo_url = "https://actix.rs/img/logo.png")]
#![doc(html_favicon_url = "https://actix.rs/favicon.ico")]
use std::marker::PhantomData;
use std::task::{Context, Poll};
use core::marker::PhantomData;
use actix_service::{
apply, dev::ApplyTransform, IntoServiceFactory, Service, ServiceFactory, Transform,
@ -36,9 +35,7 @@ where
type Error = S::Error;
type Future = Either<S::Future, Instrumented<S::Future>>;
fn poll_ready(&mut self, ctx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.inner.poll_ready(ctx)
}
actix_service::forward_ready!(inner);
fn call(&mut self, req: Req) -> Self::Future {
let span = (self.make_span)(&req);

4
actix-utils/CHANGES.md
View File

@ -1,15 +1,11 @@
# Changes
## Unreleased - 2020-xx-xx
<<<<<<< HEAD
* Upgrade `pin-project` to `1.0`.
* Update `bytes` dependency to `1`.
=======
* Use `pin-project-lite` to replace `pin-project`. [#229]
* Remove `condition`,`either`,`inflight`,`keepalive`,`oneshot`,`order`,`stream` and `time` mods. [#229]
[#229]: https://github.com/actix/actix-net/pull/229
>>>>>>> upstream/master
## 2.0.0 - 2020-08-23
* No changes from beta 1.

6
actix-utils/src/timeout.rs
View File

@ -200,7 +200,7 @@ where
#[cfg(test)]
mod tests {
use std::task::{Context, Poll};
use std::task::Poll;
use std::time::Duration;
use super::*;
@ -214,9 +214,7 @@ mod tests {
type Error = ();
type Future = LocalBoxFuture<'static, Result<(), ()>>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
actix_service::always_ready!();
fn call(&mut self, _: ()) -> Self::Future {
actix_rt::time::sleep(self.0)