actix/actix-net
1
0
Fork 0
mirror of https://github.com/fafhrd91/actix-net synced 2024-11-27 15:42:57 +01:00
Code Issues Releases Wiki Activity

remove custom NewService and cleanups

Browse Source
This commit is contained in:
Nikolay Kim 2018-08-23 15:42:34 -07:00
parent 1261ecbce0
commit a5a026b5c4
9 changed files with 92 additions and 292 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
examples/basic.rs
View File

@ -20,7 +20,7 @@ use openssl::ssl::{SslAcceptor, SslFiletype, SslMethod};
use tokio_io::{AsyncRead, AsyncWrite};
use tokio_openssl::SslAcceptorExt;
use actix_net::{IntoNewService, NewService, Server};
use actix_net::{IntoNewService, NewServiceExt, Server};
/// Simple logger service, it just prints fact of the new connections
fn logger<T: AsyncRead + AsyncWrite + fmt::Debug>(

18
examples/ssl.rs
View File

@ -15,7 +15,7 @@ use futures::{future, Future};
use openssl::ssl::{SslAcceptor, SslFiletype, SslMethod};
use tokio_io::{AsyncRead, AsyncWrite};
use actix_net::{ssl, NewService, Server};
use actix_net::{ssl, NewServiceExt, Server};
#[derive(Debug)]
struct ServiceState {
@ -46,14 +46,16 @@ fn main() {
let openssl = ssl::OpensslService::new(builder);
// server start mutiple workers, it runs supplied `Fn` in each worker.
Server::default().bind("0.0.0.0:8443", move || {
let num = num.clone();
Server::default()
.bind("0.0.0.0:8443", move || {
let num = num.clone();
// configure service
openssl.clone().and_then((service, move || {
Ok::<_, io::Error>(ServiceState { num: num.clone() })
}))
}).unwrap().start();
// configure service
openssl.clone().and_then((service, move || {
Ok::<_, io::Error>(ServiceState { num: num.clone() })
}))
}).unwrap()
.start();
sys.run();
}

8
src/lib.rs
View File

@ -45,20 +45,18 @@ extern crate webpki_roots;
use actix::Message;
/// re-export for convinience. as a note, actix-net does not use `tower_service::NewService` trait.
pub use tower_service::Service;
/// re-export for convinience
pub use tower_service::{NewService, Service};
pub(crate) mod accept;
mod server;
pub mod server_config;
mod server_service;
pub mod service;
pub mod ssl;
mod worker;
pub use server::Server;
pub use server_config::Config;
pub use service::{IntoNewService, IntoService, NewService};
pub use service::{IntoNewService, IntoService, NewServiceExt};
/// Pause accepting incoming connections
///

96
src/server.rs
View File

@ -17,10 +17,9 @@ use actix::{
};
use super::accept::{AcceptLoop, AcceptNotify, Command};
use super::server_config::{Config, ServerConfig};
use super::server_service::{ServerNewService, ServerServiceFactory};
use super::service::NewService;
use super::worker::{Conn, StopWorker, Worker, WorkerClient};
use super::NewService;
use super::{PauseServer, ResumeServer, StopServer, Token};
pub(crate) enum ServerCommand {
@ -28,11 +27,10 @@ pub(crate) enum ServerCommand {
}
/// Server
pub struct Server<C = ServerConfig> {
config: C,
pub struct Server {
threads: usize,
workers: Vec<(usize, Addr<Worker>)>,
services: Vec<Box<ServerServiceFactory<C> + Send>>,
services: Vec<Box<ServerServiceFactory + Send>>,
sockets: Vec<(Token, net::TcpListener)>,
accept: AcceptLoop,
exit: bool,
@ -43,17 +41,16 @@ pub struct Server<C = ServerConfig> {
maxconnrate: usize,
}
impl Default for Server<ServerConfig> {
impl Default for Server {
fn default() -> Self {
Self::new(ServerConfig::default())
Self::new()
}
}
impl<C: Config> Server<C> {
impl Server {
/// Create new Server instance
pub fn new(config: C) -> Server<C> {
pub fn new() -> Server {
Server {
config,
threads: num_cpus::get(),
workers: Vec::new(),
services: Vec::new(),
@ -83,10 +80,7 @@ impl<C: Config> Server<C> {
/// reached for each worker.
///
/// By default max connections is set to a 100k.
pub fn maxconn(mut self, num: usize) -> Self
where
C: AsMut<Connections>,
{
pub fn maxconn(mut self, num: usize) -> Self {
self.maxconn = num;
self
}
@ -160,9 +154,9 @@ impl<C: Config> Server<C> {
/// .handler("/static", fs::StaticFiles::new(".").unwrap());
/// }
/// ```
pub fn configure<F>(self, cfg: F) -> Server<C>
pub fn configure<F>(self, cfg: F) -> Server
where
F: Fn(Server<C>) -> Server<C>,
F: Fn(Server) -> Server,
{
cfg(self)
}
@ -172,7 +166,7 @@ impl<C: Config> Server<C> {
where
F: Fn() -> N + Clone + Send + 'static,
U: net::ToSocketAddrs,
N: NewService<Request = TcpStream, Response = (), Config = C, InitError = io::Error> + 'static,
N: NewService<Request = TcpStream, Response = (), InitError = io::Error> + 'static,
N::Service: 'static,
N::Future: 'static,
N::Error: fmt::Display,
@ -189,13 +183,13 @@ impl<C: Config> Server<C> {
pub fn listen<F, N>(mut self, lst: net::TcpListener, factory: F) -> Self
where
F: Fn() -> N + Clone + Send + 'static,
N: NewService<Request = TcpStream, Response = (), Config = C, InitError = io::Error> + 'static,
N: NewService<Request = TcpStream, Response = (), InitError = io::Error> + 'static,
N::Service: 'static,
N::Future: 'static,
N::Error: fmt::Display,
{
let token = Token(self.services.len());
self.services.push(ServerNewService::create(factory, self.config.clone()));
self.services.push(ServerNewService::create(factory));
self.sockets.push((token, lst));
self
}
@ -229,7 +223,7 @@ impl<C: Config> Server<C> {
}
/// Starts Server Actor and returns its address
pub fn start(mut self) -> Addr<Server<C>> {
pub fn start(mut self) -> Addr<Server> {
if self.sockets.is_empty() {
panic!("Service should have at least one bound socket");
} else {
@ -281,7 +275,7 @@ impl<C: Config> Server<C> {
let (tx, rx) = unbounded::<Conn>();
let conns = Connections::new(notify, self.maxconn, self.maxconnrate);
let worker = WorkerClient::new(idx, tx, conns.clone());
let services: Vec<Box<ServerServiceFactory<C> + Send>> =
let services: Vec<Box<ServerServiceFactory + Send>> =
self.services.iter().map(|v| v.clone_factory()).collect();
let addr = Arbiter::start(move |ctx: &mut Context<_>| {
@ -293,14 +287,14 @@ impl<C: Config> Server<C> {
}
}
impl<C: Config> Actor for Server<C> {
impl Actor for Server {
type Context = Context<Self>;
}
/// Signals support
/// Handle `SIGINT`, `SIGTERM`, `SIGQUIT` signals and stop actix system
/// message to `System` actor.
impl<C: Config> Handler<signal::Signal> for Server<C> {
impl Handler<signal::Signal> for Server {
type Result = ();
fn handle(&mut self, msg: signal::Signal, ctx: &mut Context<Self>) {
@ -325,7 +319,7 @@ impl<C: Config> Handler<signal::Signal> for Server<C> {
}
}
impl<C: Config> Handler<PauseServer> for Server<C> {
impl Handler<PauseServer> for Server {
type Result = ();
fn handle(&mut self, _: PauseServer, _: &mut Context<Self>) {
@ -333,7 +327,7 @@ impl<C: Config> Handler<PauseServer> for Server<C> {
}
}
impl<C: Config> Handler<ResumeServer> for Server<C> {
impl Handler<ResumeServer> for Server {
type Result = ();
fn handle(&mut self, _: ResumeServer, _: &mut Context<Self>) {
@ -341,7 +335,7 @@ impl<C: Config> Handler<ResumeServer> for Server<C> {
}
}
impl<C: Config> Handler<StopServer> for Server<C> {
impl Handler<StopServer> for Server {
type Result = Response<(), ()>;
fn handle(&mut self, msg: StopServer, ctx: &mut Context<Self>) -> Self::Result {
@ -396,7 +390,7 @@ impl<C: Config> Handler<StopServer> for Server<C> {
}
/// Commands from accept threads
impl<C: Config> StreamHandler<ServerCommand, ()> for Server<C> {
impl StreamHandler<ServerCommand, ()> for Server {
fn finished(&mut self, _: &mut Context<Self>) {}
fn handle(&mut self, msg: ServerCommand, _: &mut Context<Self>) {
@ -461,16 +455,6 @@ impl Connections {
pub(crate) fn available(&self) -> bool {
self.0.available()
}
/// Report opened connection
pub fn connection(&self) -> ConnectionTag {
ConnectionTag::new(self.0.clone())
}
/// Report rate connection, rate is usually ssl handshake
pub fn connection_rate(&self) -> ConnectionRateTag {
ConnectionRateTag::new(self.0.clone())
}
}
#[derive(Default)]
@ -506,44 +490,6 @@ impl ConnectionsInner {
}
}
/// Type responsible for max connection stat.
///
/// Max connections stat get updated on drop.
pub struct ConnectionTag(Arc<ConnectionsInner>);
impl ConnectionTag {
fn new(inner: Arc<ConnectionsInner>) -> Self {
inner.conn.fetch_add(1, Ordering::Relaxed);
ConnectionTag(inner)
}
}
impl Drop for ConnectionTag {
fn drop(&mut self) {
let conn = self.0.conn.fetch_sub(1, Ordering::Relaxed);
self.0.notify_maxconn(conn);
}
}
/// Type responsible for max connection rate stat.
///
/// Max connections rate stat get updated on drop.
pub struct ConnectionRateTag(Arc<ConnectionsInner>);
impl ConnectionRateTag {
fn new(inner: Arc<ConnectionsInner>) -> Self {
inner.connrate.fetch_add(1, Ordering::Relaxed);
ConnectionRateTag(inner)
}
}
impl Drop for ConnectionRateTag {
fn drop(&mut self) {
let connrate = self.0.connrate.fetch_sub(1, Ordering::Relaxed);
self.0.notify_maxconnrate(connrate);
}
}
fn bind_addr<S: net::ToSocketAddrs>(addr: S) -> io::Result<Vec<net::TcpListener>> {
let mut err = None;
let mut succ = false;

83
src/server_config.rs
View File

@ -1,83 +0,0 @@
//! Default server config
use std::sync::{atomic::AtomicUsize, Arc};
pub trait Config: Send + Clone + Default + 'static {
fn fork(&self) -> Self {
self.clone()
}
}
#[derive(Clone, Default)]
pub struct ServerConfig {
conn: ConnectionsConfig,
ssl: SslConfig,
}
impl Config for ServerConfig {
fn fork(&self) -> Self {
ServerConfig {
conn: self.conn.fork(),
ssl: self.ssl.fork(),
}
}
}
impl AsRef<ConnectionsConfig> for ServerConfig {
fn as_ref(&self) -> &ConnectionsConfig {
&self.conn
}
}
impl AsRef<SslConfig> for ServerConfig {
fn as_ref(&self) -> &SslConfig {
&self.ssl
}
}
#[derive(Clone)]
pub struct ConnectionsConfig {
max_connections: usize,
num_connections: Arc<AtomicUsize>,
}
impl Default for ConnectionsConfig {
fn default() -> Self {
ConnectionsConfig {
max_connections: 102_400,
num_connections: Arc::new(AtomicUsize::new(0)),
}
}
}
impl Config for ConnectionsConfig {
fn fork(&self) -> Self {
ConnectionsConfig {
max_connections: self.max_connections,
num_connections: Arc::new(AtomicUsize::new(0)),
}
}
}
#[derive(Clone)]
pub struct SslConfig {
max_handshakes: usize,
num: Arc<AtomicUsize>,
}
impl Default for SslConfig {
fn default() -> Self {
SslConfig {
max_handshakes: 256,
num: Arc::new(AtomicUsize::new(0)),
}
}
}
impl Config for SslConfig {
fn fork(&self) -> Self {
SslConfig {
max_handshakes: self.max_handshakes,
num: Arc::new(AtomicUsize::new(0)),
}
}
}

33
src/server_service.rs
View File

@ -8,7 +8,7 @@ use futures::{future, Future, Poll};
use tokio_reactor::Handle;
use tokio_tcp::TcpStream;
use super::{Config, NewService, Service};
use super::{NewService, Service};
pub(crate) type BoxedServerService = Box<
Service<
@ -56,42 +56,42 @@ where
}
}
pub(crate) struct ServerNewService<F, T, C> where F: Fn() -> T + Send + Clone {
pub(crate) struct ServerNewService<F, T>
where
F: Fn() -> T + Send + Clone,
{
inner: F,
config: C,
counter: Arc<AtomicUsize>,
}
impl<F, T, C: Config> ServerNewService<F, T, C>
impl<F, T> ServerNewService<F, T>
where
F: Fn() -> T + Send + Clone + 'static,
T: NewService<Request = TcpStream, Response = (), Config = C, InitError = io::Error> + 'static,
T: NewService<Request = TcpStream, Response = (), InitError = io::Error> + 'static,
T::Service: 'static,
T::Future: 'static,
T::Error: fmt::Display,
{
pub(crate) fn create(inner: F, config: C) -> Box<ServerServiceFactory<C> + Send> {
pub(crate) fn create(inner: F) -> Box<ServerServiceFactory + Send> {
Box::new(Self {
inner,
config,
counter: Arc::new(AtomicUsize::new(0)),
})
}
}
pub trait ServerServiceFactory<C> {
pub trait ServerServiceFactory {
fn counter(&self) -> Arc<AtomicUsize>;
fn clone_factory(&self) -> Box<ServerServiceFactory<C> + Send>;
fn clone_factory(&self) -> Box<ServerServiceFactory + Send>;
fn create(&self) -> Box<Future<Item = BoxedServerService, Error = ()>>;
}
impl<F, T, C: Config> ServerServiceFactory<C> for ServerNewService<F, T, C>
impl<F, T> ServerServiceFactory for ServerNewService<F, T>
where
F: Fn() -> T + Send + Clone + 'static,
T: NewService<Request = TcpStream, Response = (), Config = C, InitError = io::Error>
+ 'static,
T: NewService<Request = TcpStream, Response = (), InitError = io::Error> + 'static,
T::Service: 'static,
T::Future: 'static,
T::Error: fmt::Display,
@ -100,10 +100,9 @@ where
self.counter.clone()
}
fn clone_factory(&self) -> Box<ServerServiceFactory<C> + Send> {
fn clone_factory(&self) -> Box<ServerServiceFactory + Send> {
Box::new(Self {
inner: self.inner.clone(),
config: self.config.fork(),
counter: Arc::new(AtomicUsize::new(0)),
})
}
@ -112,7 +111,7 @@ where
let counter = self.counter.clone();
Box::new(
(self.inner)()
.new_service(self.config.clone())
.new_service()
.map_err(|_| ())
.map(move |inner| {
let service: BoxedServerService =
@ -123,12 +122,12 @@ where
}
}
impl<C> ServerServiceFactory<C> for Box<ServerServiceFactory<C>> {
impl ServerServiceFactory for Box<ServerServiceFactory> {
fn counter(&self) -> Arc<AtomicUsize> {
self.as_ref().counter()
}
fn clone_factory(&self) -> Box<ServerServiceFactory<C> + Send> {
fn clone_factory(&self) -> Box<ServerServiceFactory + Send> {
self.as_ref().clone_factory()
}

118
src/service.rs
View File

@ -3,44 +3,9 @@ use std::marker;
use std::rc::Rc;
use futures::{future, future::FutureResult, Async, Future, IntoFuture, Poll};
use tower_service::Service;
/// Creates new `Service` values.
///
/// Acts as a service factory. This is useful for cases where new `Service`
/// values must be produced. One case is a TCP servier listener. The listner
/// accepts new TCP streams, obtains a new `Service` value using the
/// `NewService` trait, and uses that new `Service` value to process inbound
/// requests on that new TCP stream.
pub trait NewService {
/// Requests handled by the service
type Request;
/// Responses given by the service
type Response;
/// Errors produced by the service
type Error;
/// The `Service` value created by this factory
type Service: Service<
Request = Self::Request,
Response = Self::Response,
Error = Self::Error,
>;
/// Pipeline configuration
type Config: Clone;
/// Errors produced while building a service.
type InitError;
/// The future of the `Service` instance.
type Future: Future<Item = Self::Service, Error = Self::InitError>;
/// Create and return a new service value asynchronously.
fn new_service(&self, Self::Config) -> Self::Future;
use tower_service::{NewService, Service};
pub trait NewServiceExt: NewService {
fn and_then<F, B>(self, new_service: F) -> AndThenNewService<Self, B>
where
Self: Sized,
@ -48,7 +13,6 @@ pub trait NewService {
B: NewService<
Request = Self::Response,
Error = Self::Error,
Config = Self::Config,
InitError = Self::InitError,
>,
{
@ -72,6 +36,8 @@ pub trait NewService {
}
}
impl<T: NewService> NewServiceExt for T {}
/// Trait for types that can be converted to a Service
pub trait IntoService<T>
where
@ -163,7 +129,7 @@ where
}
}
pub struct FnNewService<F, Req, Resp, Err, IErr, Fut, Cfg>
pub struct FnNewService<F, Req, Resp, Err, IErr, Fut>
where
F: Fn(Req) -> Fut,
Fut: IntoFuture<Item = Resp, Error = Err>,
@ -173,10 +139,9 @@ where
resp: marker::PhantomData<Resp>,
err: marker::PhantomData<Err>,
ierr: marker::PhantomData<IErr>,
cfg: marker::PhantomData<Cfg>,
}
impl<F, Req, Resp, Err, IErr, Fut, Cfg> FnNewService<F, Req, Resp, Err, IErr, Fut, Cfg>
impl<F, Req, Resp, Err, IErr, Fut> FnNewService<F, Req, Resp, Err, IErr, Fut>
where
F: Fn(Req) -> Fut + Clone,
Fut: IntoFuture<Item = Resp, Error = Err>,
@ -188,43 +153,39 @@ where
resp: marker::PhantomData,
err: marker::PhantomData,
ierr: marker::PhantomData,
cfg: marker::PhantomData,
}
}
}
impl<F, Req, Resp, Err, IErr, Fut, Cfg> NewService for FnNewService<F, Req, Resp, Err, IErr, Fut, Cfg>
impl<F, Req, Resp, Err, IErr, Fut> NewService for FnNewService<F, Req, Resp, Err, IErr, Fut>
where
F: Fn(Req) -> Fut + Clone,
Fut: IntoFuture<Item = Resp, Error = Err>,
Cfg: Clone,
{
type Request = Req;
type Response = Resp;
type Error = Err;
type Service = FnService<F, Req, Resp, Err, Fut>;
type Config = Cfg;
type InitError = IErr;
type Future = FutureResult<Self::Service, Self::InitError>;
fn new_service(&self, _: Cfg) -> Self::Future {
fn new_service(&self) -> Self::Future {
future::ok(FnService::new(self.f.clone()))
}
}
impl<F, Req, Resp, Err, IErr, Fut, Cfg> IntoNewService<FnNewService<F, Req, Resp, Err, IErr, Fut, Cfg>>
impl<F, Req, Resp, Err, IErr, Fut> IntoNewService<FnNewService<F, Req, Resp, Err, IErr, Fut>>
for F
where
F: Fn(Req) -> Fut + Clone + 'static,
Fut: IntoFuture<Item = Resp, Error = Err>,
Cfg: Clone,
{
fn into_new_service(self) -> FnNewService<F, Req, Resp, Err, IErr, Fut, Cfg> {
fn into_new_service(self) -> FnNewService<F, Req, Resp, Err, IErr, Fut> {
FnNewService::new(self)
}
}
impl<F, Req, Resp, Err, IErr, Fut, Cfg> Clone for FnNewService<F, Req, Resp, Err, IErr, Fut, Cfg>
impl<F, Req, Resp, Err, IErr, Fut> Clone for FnNewService<F, Req, Resp, Err, IErr, Fut>
where
F: Fn(Req) -> Fut + Clone,
Fut: IntoFuture<Item = Resp, Error = Err>,
@ -282,7 +243,7 @@ where
}
/// `NewService` for state and handler functions
pub struct FnStateNewService<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2, Cfg> {
pub struct FnStateNewService<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2> {
f: F1,
state: F2,
s: marker::PhantomData<S>,
@ -292,11 +253,10 @@ pub struct FnStateNewService<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2, Cfg>
err2: marker::PhantomData<Err2>,
fut1: marker::PhantomData<Fut1>,
fut2: marker::PhantomData<Fut2>,
cfg: marker::PhantomData<Cfg>,
}
impl<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2, Cfg>
FnStateNewService<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2, Cfg>
impl<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2>
FnStateNewService<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2>
{
fn new(f: F1, state: F2) -> Self {
FnStateNewService {
@ -309,13 +269,12 @@ impl<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2, Cfg>
err2: marker::PhantomData,
fut1: marker::PhantomData,
fut2: marker::PhantomData,
cfg: marker::PhantomData,
}
}
}
impl<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2, Cfg> NewService
for FnStateNewService<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2, Cfg>
impl<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2> NewService
for FnStateNewService<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2>
where
S: 'static,
F1: Fn(&mut S, Req) -> Fut1 + Clone + 'static,
@ -326,17 +285,15 @@ where
Resp: 'static,
Err1: 'static,
Err2: 'static,
Cfg: Clone,
{
type Request = Req;
type Response = Resp;
type Error = Err1;
type Service = FnStateService<S, F1, Req, Resp, Err1, Fut1>;
type Config = Cfg;
type InitError = Err2;
type Future = Box<Future<Item = Self::Service, Error = Self::InitError>>;
fn new_service(&self, _: Cfg) -> Self::Future {
fn new_service(&self) -> Self::Future {
let f = self.f.clone();
Box::new(
(self.state)()
@ -346,9 +303,8 @@ where
}
}
impl<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2, Cfg>
IntoNewService<FnStateNewService<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2, Cfg>>
for (F1, F2)
impl<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2>
IntoNewService<FnStateNewService<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2>> for (F1, F2)
where
S: 'static,
F1: Fn(&mut S, Req) -> Fut1 + Clone + 'static,
@ -359,17 +315,16 @@ where
Resp: 'static,
Err1: 'static,
Err2: 'static,
Cfg: Clone,
{
fn into_new_service(
self,
) -> FnStateNewService<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2, Cfg> {
) -> FnStateNewService<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2> {
FnStateNewService::new(self.0, self.1)
}
}
impl<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2, Cfg> Clone
for FnStateNewService<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2, Cfg>
impl<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2> Clone
for FnStateNewService<S, F1, F2, Req, Resp, Err1, Err2, Fut1, Fut2>
where
F1: Fn(&mut S, Req) -> Fut1 + Clone + 'static,
F2: Fn() -> Fut2 + Clone,
@ -415,11 +370,9 @@ where
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
match self.a.poll_ready() {
Ok(Async::Ready(_)) => {
self.b.borrow_mut().poll_ready()
},
Ok(Async::Ready(_)) => self.b.borrow_mut().poll_ready(),
Ok(Async::NotReady) => Ok(Async::NotReady),
Err(err) => Err(err.into())
Err(err) => Err(err.into()),
}
}
@ -474,7 +427,7 @@ where
self.poll()
}
Ok(Async::NotReady) => Ok(Async::NotReady),
Err(err) => Err(err.into())
Err(err) => Err(err.into()),
}
}
}
@ -501,11 +454,7 @@ where
impl<A, B> NewService for AndThenNewService<A, B>
where
A: NewService<
Response = B::Request,
Config = B::Config,
InitError = B::InitError,
>,
A: NewService<Response = B::Request, InitError = B::InitError>,
A::Error: Into<B::Error>,
B: NewService,
{
@ -514,12 +463,11 @@ where
type Error = B::Error;
type Service = AndThen<A::Service, B::Service>;
type Config = A::Config;
type InitError = A::InitError;
type Future = AndThenNewServiceFuture<A, B>;
fn new_service(&self, cfg: A::Config) -> Self::Future {
AndThenNewServiceFuture::new(self.a.new_service(cfg.clone()), self.b.new_service(cfg))
fn new_service(&self) -> Self::Future {
AndThenNewServiceFuture::new(self.a.new_service(), self.b.new_service())
}
}
@ -712,12 +660,11 @@ where
type Error = E;
type Service = MapErr<A::Service, F, E>;
type Config = A::Config;
type InitError = A::InitError;
type Future = MapErrNewServiceFuture<A, F, E>;
fn new_service(&self, cfg: Self::Config) -> Self::Future {
MapErrNewServiceFuture::new(self.a.new_service(cfg), self.f.clone())
fn new_service(&self) -> Self::Future {
MapErrNewServiceFuture::new(self.a.new_service(), self.f.clone())
}
}
@ -803,12 +750,11 @@ where
type Error = A::Error;
type Service = A::Service;
type Config = A::Config;
type InitError = E;
type Future = MapInitErrFuture<A, F, E>;
fn new_service(&self, cfg: Self::Config) -> Self::Future {
MapInitErrFuture::new(self.a.new_service(cfg), self.f.clone())
fn new_service(&self) -> Self::Future {
MapInitErrFuture::new(self.a.new_service(), self.f.clone())
}
}

18
src/ssl/openssl.rs
View File

@ -7,25 +7,22 @@ use openssl::ssl::{AlpnError, SslAcceptor, SslAcceptorBuilder};
use tokio_io::{AsyncRead, AsyncWrite};
use tokio_openssl::{AcceptAsync, SslAcceptorExt, SslStream};
use server_config::SslConfig;
use {NewService, Service};
/// Support `SSL` connections via openssl package
///
/// `alpn` feature enables `OpensslAcceptor` type
pub struct OpensslService<T, Cfg> {
pub struct OpensslService<T> {
acceptor: SslAcceptor,
io: PhantomData<T>,
cfg: PhantomData<Cfg>,
}
impl<T, Cfg> OpensslService<T, Cfg> {
impl<T> OpensslService<T> {
/// Create default `OpensslService`
pub fn new(builder: SslAcceptorBuilder) -> Self {
OpensslService {
acceptor: builder.build(),
io: PhantomData,
cfg: PhantomData,
}
}
@ -46,32 +43,27 @@ impl<T, Cfg> OpensslService<T, Cfg> {
Ok(OpensslService {
acceptor: builder.build(),
io: PhantomData,
cfg: PhantomData,
})
}
}
impl<T: AsyncRead + AsyncWrite, Cfg> Clone for OpensslService<T, Cfg> {
impl<T: AsyncRead + AsyncWrite> Clone for OpensslService<T> {
fn clone(&self) -> Self {
Self {
acceptor: self.acceptor.clone(),
io: PhantomData,
cfg: PhantomData,
}
}
}
impl<T: AsyncRead + AsyncWrite, Cfg: Clone + AsRef<SslConfig>> NewService
for OpensslService<T, Cfg>
{
impl<T: AsyncRead + AsyncWrite> NewService for OpensslService<T> {
type Request = T;
type Response = SslStream<T>;
type Error = io::Error;
type Service = OpensslAcceptor<T>;
type Config = Cfg;
type InitError = io::Error;
type Future = FutureResult<Self::Service, io::Error>;
fn new_service(&self, _: Self::Config) -> Self::Future {
fn new_service(&self) -> Self::Future {
future::ok(OpensslAcceptor {
acceptor: self.acceptor.clone(),
io: PhantomData,

4
src/worker.rs
View File

@ -69,8 +69,8 @@ impl Actor for Worker {
}
impl Worker {
pub(crate) fn new<C: 'static>(
ctx: &mut Context<Self>, services: Vec<Box<ServerServiceFactory<C> + Send>>,
pub(crate) fn new(
ctx: &mut Context<Self>, services: Vec<Box<ServerServiceFactory + Send>>,
) -> Self {
let wrk = Worker {
services: Vec::new(),