mirror of
https://github.com/fafhrd91/actix-web
synced 2025-07-08 11:56:32 +02:00
Compare commits
32 Commits
Author | SHA1 | Date | |
---|---|---|---|
410204a41e | |||
29fe995a02 | |||
741b3fb1c0 | |||
dba86fbbf8 | |||
c29501fcf3 | |||
9a3321b153 | |||
f13a0925f7 | |||
8886672ae6 | |||
8f20f69559 | |||
9b9599500a | |||
a4b81a256c | |||
38235c14bb | |||
bf9269de9a | |||
bb34df8c1b | |||
1ac018dc79 | |||
0e3d1068da | |||
60144a3cb8 | |||
6c25becd3f | |||
dc19a9f862 | |||
67961f8a36 | |||
3d51aa7115 | |||
a8a831a098 | |||
4b16af29c7 | |||
c15e4b92a8 | |||
e1418018c6 | |||
ff914f79fc | |||
3a133e3974 | |||
0b0d14d1ea | |||
099ebbfaa3 | |||
afe15ba44f | |||
4eabf3994a | |||
e32961a897 |
49
CHANGES.md
49
CHANGES.md
@ -1,5 +1,54 @@
|
||||
# Changes
|
||||
|
||||
## [0.2.3] - 2018-11-17
|
||||
|
||||
### Added
|
||||
|
||||
* Framed::is_write_buf_empty() checks if write buffer is flushed
|
||||
|
||||
## [0.2.2] - 2018-11-14
|
||||
|
||||
### Added
|
||||
|
||||
* Add low/high caps to Framed
|
||||
|
||||
### Changed
|
||||
|
||||
* Refactor Connector and Resolver services
|
||||
|
||||
|
||||
### Fixed
|
||||
|
||||
* Fix wrong service to socket binding
|
||||
|
||||
|
||||
## [0.2.0] - 2018-11-08
|
||||
|
||||
### Added
|
||||
|
||||
* Timeout service
|
||||
|
||||
* Added ServiceConfig and ServiceRuntime for server service configuration
|
||||
|
||||
|
||||
### Changed
|
||||
|
||||
* Connector has been refactored
|
||||
|
||||
* timer and LowResTimer renamed to time and LowResTime
|
||||
|
||||
* Refactored `Server::configure()` method
|
||||
|
||||
|
||||
## [0.1.1] - 2018-10-10
|
||||
|
||||
### Changed
|
||||
|
||||
- Set actix min version - 0.7.5
|
||||
|
||||
- Set trust-dns min version
|
||||
|
||||
|
||||
## [0.1.0] - 2018-10-08
|
||||
|
||||
* Initial impl
|
||||
|
@ -1,13 +1,13 @@
|
||||
[package]
|
||||
name = "actix-net"
|
||||
version = "0.1.0"
|
||||
version = "0.2.3"
|
||||
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
||||
description = "Actix net - framework for the compisible network services for Rust (experimental)"
|
||||
readme = "README.md"
|
||||
keywords = ["network", "framework", "async", "futures"]
|
||||
homepage = "https://actix.rs"
|
||||
repository = "https://github.com/actix/actix-net.git"
|
||||
documentation = "https://actix.rs/api/actix-net/stable/actix_net/"
|
||||
documentation = "https://docs.rs/actix-net/"
|
||||
categories = ["network-programming", "asynchronous"]
|
||||
license = "MIT/Apache-2.0"
|
||||
exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
|
||||
@ -39,7 +39,7 @@ rust-tls = ["rustls", "tokio-rustls", "webpki", "webpki-roots"]
|
||||
cell = []
|
||||
|
||||
[dependencies]
|
||||
actix = "0.7.0"
|
||||
actix = "0.7.6"
|
||||
|
||||
log = "0.4"
|
||||
num_cpus = "1.0"
|
||||
@ -58,7 +58,8 @@ tokio-timer = "0.2"
|
||||
tokio-reactor = "0.1"
|
||||
tokio-current-thread = "0.1"
|
||||
tower-service = "0.1"
|
||||
trust-dns-resolver = "0.10.0-alpha.2"
|
||||
trust-dns-proto = "^0.5.0"
|
||||
trust-dns-resolver = "^0.10.0"
|
||||
|
||||
# native-tls
|
||||
native-tls = { version="0.2", optional = true }
|
||||
|
@ -24,7 +24,8 @@ struct ServiceState {
|
||||
}
|
||||
|
||||
fn service<T: AsyncRead + AsyncWrite>(
|
||||
st: &mut ServiceState, _: T,
|
||||
st: &mut ServiceState,
|
||||
_: T,
|
||||
) -> impl Future<Item = (), Error = ()> {
|
||||
let num = st.num.fetch_add(1, Ordering::Relaxed);
|
||||
println!("got ssl connection {:?}", num);
|
||||
|
33
src/codec/bcodec.rs
Normal file
33
src/codec/bcodec.rs
Normal file
@ -0,0 +1,33 @@
|
||||
use std::io;
|
||||
|
||||
use bytes::{Bytes, BytesMut};
|
||||
use tokio_codec::{Decoder, Encoder};
|
||||
|
||||
/// Bytes codec.
|
||||
///
|
||||
/// Reads/Writes chunks of bytes from a stream.
|
||||
#[derive(Debug, Copy, Clone)]
|
||||
pub struct BytesCodec;
|
||||
|
||||
impl Encoder for BytesCodec {
|
||||
type Item = Bytes;
|
||||
type Error = io::Error;
|
||||
|
||||
fn encode(&mut self, item: Bytes, dst: &mut BytesMut) -> Result<(), Self::Error> {
|
||||
dst.extend_from_slice(&item[..]);
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl Decoder for BytesCodec {
|
||||
type Item = BytesMut;
|
||||
type Error = io::Error;
|
||||
|
||||
fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
|
||||
if src.is_empty() {
|
||||
Ok(None)
|
||||
} else {
|
||||
Ok(Some(src.take()))
|
||||
}
|
||||
}
|
||||
}
|
@ -11,6 +11,9 @@ use tokio_io::{AsyncRead, AsyncWrite};
|
||||
use super::framed_read::{framed_read2, framed_read2_with_buffer, FramedRead2};
|
||||
use super::framed_write::{framed_write2, framed_write2_with_buffer, FramedWrite2};
|
||||
|
||||
const LW: usize = 1024;
|
||||
const HW: usize = 8 * 1024;
|
||||
|
||||
/// A unified `Stream` and `Sink` interface to an underlying I/O object, using
|
||||
/// the `Encoder` and `Decoder` traits to encode and decode frames.
|
||||
///
|
||||
@ -45,9 +48,25 @@ where
|
||||
/// break them into separate objects, allowing them to interact more easily.
|
||||
pub fn new(inner: T, codec: U) -> Framed<T, U> {
|
||||
Framed {
|
||||
inner: framed_read2(framed_write2(Fuse(inner, codec))),
|
||||
inner: framed_read2(framed_write2(Fuse(inner, codec), LW, HW)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Same as `Framed::new()` with ability to specify write buffer low/high capacity watermarks.
|
||||
pub fn new_with_caps(inner: T, codec: U, lw: usize, hw: usize) -> Framed<T, U> {
|
||||
debug_assert!((lw < hw) && hw != 0);
|
||||
Framed {
|
||||
inner: framed_read2(framed_write2(Fuse(inner, codec), lw, hw)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Force send item
|
||||
pub fn force_send(
|
||||
&mut self,
|
||||
item: <U as Encoder>::Item,
|
||||
) -> Result<(), <U as Encoder>::Error> {
|
||||
self.inner.get_mut().force_send(item)
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, U> Framed<T, U> {
|
||||
@ -75,7 +94,12 @@ impl<T, U> Framed<T, U> {
|
||||
pub fn from_parts(parts: FramedParts<T, U>) -> Framed<T, U> {
|
||||
Framed {
|
||||
inner: framed_read2_with_buffer(
|
||||
framed_write2_with_buffer(Fuse(parts.io, parts.codec), parts.write_buf),
|
||||
framed_write2_with_buffer(
|
||||
Fuse(parts.io, parts.codec),
|
||||
parts.write_buf,
|
||||
parts.write_buf_lw,
|
||||
parts.write_buf_hw,
|
||||
),
|
||||
parts.read_buf,
|
||||
),
|
||||
}
|
||||
@ -111,6 +135,16 @@ impl<T, U> Framed<T, U> {
|
||||
&mut self.inner.get_mut().get_mut().0
|
||||
}
|
||||
|
||||
/// Check if write buffer is empty.
|
||||
pub fn is_write_buf_empty(&self) -> bool {
|
||||
self.inner.get_ref().is_empty()
|
||||
}
|
||||
|
||||
/// Check if write buffer is full.
|
||||
pub fn is_write_buf_full(&self) -> bool {
|
||||
self.inner.get_ref().is_full()
|
||||
}
|
||||
|
||||
/// Consumes the `Frame`, returning its underlying I/O stream.
|
||||
///
|
||||
/// Note that care should be taken to not tamper with the underlying stream
|
||||
@ -123,11 +157,27 @@ impl<T, U> Framed<T, U> {
|
||||
/// Consume the `Frame`, returning `Frame` with different codec.
|
||||
pub fn into_framed<U2>(self, codec: U2) -> Framed<T, U2> {
|
||||
let (inner, read_buf) = self.inner.into_parts();
|
||||
let (inner, write_buf) = inner.into_parts();
|
||||
let (inner, write_buf, lw, hw) = inner.into_parts();
|
||||
|
||||
Framed {
|
||||
inner: framed_read2_with_buffer(
|
||||
framed_write2_with_buffer(Fuse(inner.0, codec), write_buf),
|
||||
framed_write2_with_buffer(Fuse(inner.0, codec), write_buf, lw, hw),
|
||||
read_buf,
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
/// Consume the `Frame`, returning `Frame` with different codec.
|
||||
pub fn map_codec<F, U2>(self, f: F) -> Framed<T, U2>
|
||||
where
|
||||
F: Fn(U) -> U2,
|
||||
{
|
||||
let (inner, read_buf) = self.inner.into_parts();
|
||||
let (inner, write_buf, lw, hw) = inner.into_parts();
|
||||
|
||||
Framed {
|
||||
inner: framed_read2_with_buffer(
|
||||
framed_write2_with_buffer(Fuse(inner.0, f(inner.1)), write_buf, lw, hw),
|
||||
read_buf,
|
||||
),
|
||||
}
|
||||
@ -141,13 +191,15 @@ impl<T, U> Framed<T, U> {
|
||||
/// being worked with.
|
||||
pub fn into_parts(self) -> FramedParts<T, U> {
|
||||
let (inner, read_buf) = self.inner.into_parts();
|
||||
let (inner, write_buf) = inner.into_parts();
|
||||
let (inner, write_buf, write_buf_lw, write_buf_hw) = inner.into_parts();
|
||||
|
||||
FramedParts {
|
||||
io: inner.0,
|
||||
codec: inner.1,
|
||||
read_buf: read_buf,
|
||||
write_buf: write_buf,
|
||||
read_buf,
|
||||
write_buf,
|
||||
write_buf_lw,
|
||||
write_buf_hw,
|
||||
_priv: (),
|
||||
}
|
||||
}
|
||||
@ -176,7 +228,8 @@ where
|
||||
type SinkError = U::Error;
|
||||
|
||||
fn start_send(
|
||||
&mut self, item: Self::SinkItem,
|
||||
&mut self,
|
||||
item: Self::SinkItem,
|
||||
) -> StartSend<Self::SinkItem, Self::SinkError> {
|
||||
self.inner.get_mut().start_send(item)
|
||||
}
|
||||
@ -272,6 +325,12 @@ pub struct FramedParts<T, U> {
|
||||
/// A buffer with unprocessed data which are not written yet.
|
||||
pub write_buf: BytesMut,
|
||||
|
||||
/// A buffer low watermark capacity
|
||||
pub write_buf_lw: usize,
|
||||
|
||||
/// A buffer high watermark capacity
|
||||
pub write_buf_hw: usize,
|
||||
|
||||
/// This private field allows us to add additional fields in the future in a
|
||||
/// backwards compatible way.
|
||||
_priv: (),
|
||||
@ -285,6 +344,8 @@ impl<T, U> FramedParts<T, U> {
|
||||
codec,
|
||||
read_buf: BytesMut::new(),
|
||||
write_buf: BytesMut::new(),
|
||||
write_buf_lw: LW,
|
||||
write_buf_hw: HW,
|
||||
_priv: (),
|
||||
}
|
||||
}
|
||||
|
@ -191,7 +191,8 @@ where
|
||||
type SinkError = E::Error;
|
||||
|
||||
fn start_send(
|
||||
&mut self, item: Self::SinkItem,
|
||||
&mut self,
|
||||
item: Self::SinkItem,
|
||||
) -> StartSend<Self::SinkItem, Self::SinkError> {
|
||||
self.inner.get_mut().start_send(item)
|
||||
}
|
||||
|
@ -98,7 +98,8 @@ where
|
||||
type SinkError = T::SinkError;
|
||||
|
||||
fn start_send(
|
||||
&mut self, item: Self::SinkItem,
|
||||
&mut self,
|
||||
item: Self::SinkItem,
|
||||
) -> StartSend<Self::SinkItem, Self::SinkError> {
|
||||
self.inner.inner.0.start_send(item)
|
||||
}
|
||||
|
@ -16,20 +16,19 @@ pub struct FramedWrite<T, E> {
|
||||
pub struct FramedWrite2<T> {
|
||||
inner: T,
|
||||
buffer: BytesMut,
|
||||
low_watermark: usize,
|
||||
high_watermark: usize,
|
||||
}
|
||||
|
||||
const INITIAL_CAPACITY: usize = 8 * 1024;
|
||||
const BACKPRESSURE_BOUNDARY: usize = INITIAL_CAPACITY;
|
||||
|
||||
impl<T, E> FramedWrite<T, E>
|
||||
where
|
||||
T: AsyncWrite,
|
||||
E: Encoder,
|
||||
{
|
||||
/// Creates a new `FramedWrite` with the given `encoder`.
|
||||
pub fn new(inner: T, encoder: E) -> FramedWrite<T, E> {
|
||||
pub fn new(inner: T, encoder: E, lw: usize, hw: usize) -> FramedWrite<T, E> {
|
||||
FramedWrite {
|
||||
inner: framed_write2(Fuse(inner, encoder)),
|
||||
inner: framed_write2(Fuse(inner, encoder), lw, hw),
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -73,6 +72,26 @@ impl<T, E> FramedWrite<T, E> {
|
||||
pub fn encoder_mut(&mut self) -> &mut E {
|
||||
&mut self.inner.inner.1
|
||||
}
|
||||
|
||||
/// Check if write buffer is full
|
||||
pub fn is_full(&self) -> bool {
|
||||
self.inner.is_full()
|
||||
}
|
||||
|
||||
/// Check if write buffer is empty.
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.inner.is_empty()
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, E> FramedWrite<T, E>
|
||||
where
|
||||
E: Encoder,
|
||||
{
|
||||
/// Force send item
|
||||
pub fn force_send(&mut self, item: E::Item) -> Result<(), E::Error> {
|
||||
self.inner.force_send(item)
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, E> Sink for FramedWrite<T, E>
|
||||
@ -124,21 +143,34 @@ where
|
||||
|
||||
// ===== impl FramedWrite2 =====
|
||||
|
||||
pub fn framed_write2<T>(inner: T) -> FramedWrite2<T> {
|
||||
pub fn framed_write2<T>(
|
||||
inner: T,
|
||||
low_watermark: usize,
|
||||
high_watermark: usize,
|
||||
) -> FramedWrite2<T> {
|
||||
FramedWrite2 {
|
||||
inner: inner,
|
||||
buffer: BytesMut::with_capacity(INITIAL_CAPACITY),
|
||||
inner,
|
||||
low_watermark,
|
||||
high_watermark,
|
||||
buffer: BytesMut::with_capacity(high_watermark),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn framed_write2_with_buffer<T>(inner: T, mut buf: BytesMut) -> FramedWrite2<T> {
|
||||
if buf.capacity() < INITIAL_CAPACITY {
|
||||
let bytes_to_reserve = INITIAL_CAPACITY - buf.capacity();
|
||||
buf.reserve(bytes_to_reserve);
|
||||
pub fn framed_write2_with_buffer<T>(
|
||||
inner: T,
|
||||
mut buffer: BytesMut,
|
||||
low_watermark: usize,
|
||||
high_watermark: usize,
|
||||
) -> FramedWrite2<T> {
|
||||
if buffer.capacity() < high_watermark {
|
||||
let bytes_to_reserve = high_watermark - buffer.capacity();
|
||||
buffer.reserve(bytes_to_reserve);
|
||||
}
|
||||
FramedWrite2 {
|
||||
inner: inner,
|
||||
buffer: buf,
|
||||
inner,
|
||||
buffer,
|
||||
low_watermark,
|
||||
high_watermark,
|
||||
}
|
||||
}
|
||||
|
||||
@ -151,13 +183,40 @@ impl<T> FramedWrite2<T> {
|
||||
self.inner
|
||||
}
|
||||
|
||||
pub fn into_parts(self) -> (T, BytesMut) {
|
||||
(self.inner, self.buffer)
|
||||
pub fn into_parts(self) -> (T, BytesMut, usize, usize) {
|
||||
(
|
||||
self.inner,
|
||||
self.buffer,
|
||||
self.low_watermark,
|
||||
self.high_watermark,
|
||||
)
|
||||
}
|
||||
|
||||
pub fn get_mut(&mut self) -> &mut T {
|
||||
&mut self.inner
|
||||
}
|
||||
|
||||
pub fn is_full(&self) -> bool {
|
||||
self.buffer.len() >= self.high_watermark
|
||||
}
|
||||
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.buffer.is_empty()
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> FramedWrite2<T>
|
||||
where
|
||||
T: Encoder,
|
||||
{
|
||||
pub fn force_send(&mut self, item: T::Item) -> Result<(), T::Error> {
|
||||
let len = self.buffer.len();
|
||||
if len < self.low_watermark {
|
||||
self.buffer.reserve(self.high_watermark - len)
|
||||
}
|
||||
self.inner.encode(item, &mut self.buffer)?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> Sink for FramedWrite2<T>
|
||||
@ -168,18 +227,16 @@ where
|
||||
type SinkError = T::Error;
|
||||
|
||||
fn start_send(&mut self, item: T::Item) -> StartSend<T::Item, T::Error> {
|
||||
// If the buffer is already over 8KiB, then attempt to flush it. If after
|
||||
// flushing it's *still* over 8KiB, then apply backpressure (reject the
|
||||
// send).
|
||||
if self.buffer.len() >= BACKPRESSURE_BOUNDARY {
|
||||
try!(self.poll_complete());
|
||||
|
||||
if self.buffer.len() >= BACKPRESSURE_BOUNDARY {
|
||||
return Ok(AsyncSink::NotReady(item));
|
||||
}
|
||||
// Check the buffer capacity
|
||||
let len = self.buffer.len();
|
||||
if len >= self.high_watermark {
|
||||
return Ok(AsyncSink::NotReady(item));
|
||||
}
|
||||
if len < self.low_watermark {
|
||||
self.buffer.reserve(self.high_watermark - len)
|
||||
}
|
||||
|
||||
try!(self.inner.encode(item, &mut self.buffer));
|
||||
self.inner.encode(item, &mut self.buffer)?;
|
||||
|
||||
Ok(AsyncSink::Ready)
|
||||
}
|
||||
|
@ -12,12 +12,14 @@
|
||||
|
||||
#![deny(missing_docs, missing_debug_implementations, warnings)]
|
||||
|
||||
mod bcodec;
|
||||
mod framed;
|
||||
mod framed2;
|
||||
// mod framed2;
|
||||
mod framed_read;
|
||||
mod framed_write;
|
||||
|
||||
pub use self::bcodec::BytesCodec;
|
||||
pub use self::framed::{Framed, FramedParts};
|
||||
pub use self::framed2::{Framed2, FramedParts2};
|
||||
// pub use self::framed2::{Framed2, FramedParts2};
|
||||
pub use self::framed_read::FramedRead;
|
||||
pub use self::framed_write::FramedWrite;
|
||||
|
321
src/connector.rs
321
src/connector.rs
@ -1,6 +1,8 @@
|
||||
use std::collections::VecDeque;
|
||||
use std::io;
|
||||
use std::net::SocketAddr;
|
||||
use std::marker::PhantomData;
|
||||
use std::net::{IpAddr, SocketAddr};
|
||||
use std::time::Duration;
|
||||
use std::{fmt, io};
|
||||
|
||||
use futures::{
|
||||
future::{ok, FutureResult},
|
||||
@ -10,41 +12,115 @@ use tokio_tcp::{ConnectFuture, TcpStream};
|
||||
use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
|
||||
use trust_dns_resolver::system_conf::read_system_conf;
|
||||
|
||||
use super::resolver::{HostAware, Resolver, ResolverError, ResolverFuture};
|
||||
use super::resolver::{RequestHost, ResolveError, Resolver, ResolverFuture};
|
||||
use super::service::{NewService, Service};
|
||||
|
||||
/// Port of the request
|
||||
pub trait RequestPort {
|
||||
fn port(&self) -> u16;
|
||||
}
|
||||
|
||||
// #[derive(Fail, Debug)]
|
||||
#[derive(Debug)]
|
||||
pub enum ConnectorError {
|
||||
/// Failed to resolve the hostname
|
||||
// #[fail(display = "Failed resolving hostname: {}", _0)]
|
||||
Resolver(ResolverError),
|
||||
Resolver(ResolveError),
|
||||
|
||||
/// Not dns records
|
||||
// #[fail(display = "Invalid input: {}", _0)]
|
||||
/// No dns records
|
||||
// #[fail(display = "No dns records found for the input")]
|
||||
NoRecords,
|
||||
|
||||
/// Connecting took too long
|
||||
// #[fail(display = "Timeout out while establishing connection")]
|
||||
Timeout,
|
||||
|
||||
/// Invalid input
|
||||
InvalidInput,
|
||||
|
||||
/// Connection io error
|
||||
// #[fail(display = "{}", _0)]
|
||||
IoError(io::Error),
|
||||
}
|
||||
|
||||
impl From<ResolverError> for ConnectorError {
|
||||
fn from(err: ResolverError) -> Self {
|
||||
impl From<ResolveError> for ConnectorError {
|
||||
fn from(err: ResolveError) -> Self {
|
||||
ConnectorError::Resolver(err)
|
||||
}
|
||||
}
|
||||
|
||||
pub struct ConnectionInfo {
|
||||
impl From<io::Error> for ConnectorError {
|
||||
fn from(err: io::Error) -> Self {
|
||||
ConnectorError::IoError(err)
|
||||
}
|
||||
}
|
||||
|
||||
/// Connect request
|
||||
#[derive(Eq, PartialEq, Debug, Hash)]
|
||||
pub struct Connect {
|
||||
pub host: String,
|
||||
pub addr: SocketAddr,
|
||||
pub port: u16,
|
||||
pub timeout: Duration,
|
||||
}
|
||||
|
||||
pub struct Connector<T = String> {
|
||||
resolver: Resolver<T>,
|
||||
impl Connect {
|
||||
/// Create new `Connect` instance.
|
||||
pub fn new<T: AsRef<str>>(host: T, port: u16) -> Connect {
|
||||
Connect {
|
||||
port,
|
||||
host: host.as_ref().to_owned(),
|
||||
timeout: Duration::from_secs(1),
|
||||
}
|
||||
}
|
||||
|
||||
/// Create `Connect` instance by spliting the string by ':' and convert the second part to u16
|
||||
pub fn with<T: AsRef<str>>(host: T) -> Result<Connect, ConnectorError> {
|
||||
let mut parts_iter = host.as_ref().splitn(2, ':');
|
||||
let host = parts_iter.next().ok_or(ConnectorError::InvalidInput)?;
|
||||
let port_str = parts_iter.next().unwrap_or("");
|
||||
let port = port_str
|
||||
.parse::<u16>()
|
||||
.map_err(|_| ConnectorError::InvalidInput)?;
|
||||
Ok(Connect {
|
||||
port,
|
||||
host: host.to_owned(),
|
||||
timeout: Duration::from_secs(1),
|
||||
})
|
||||
}
|
||||
|
||||
/// Set connect timeout
|
||||
///
|
||||
/// By default timeout is set to a 1 second.
|
||||
pub fn timeout(mut self, timeout: Duration) -> Connect {
|
||||
self.timeout = timeout;
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: HostAware> Default for Connector<T> {
|
||||
impl RequestHost for Connect {
|
||||
fn host(&self) -> &str {
|
||||
&self.host
|
||||
}
|
||||
}
|
||||
|
||||
impl RequestPort for Connect {
|
||||
fn port(&self) -> u16 {
|
||||
self.port
|
||||
}
|
||||
}
|
||||
|
||||
impl fmt::Display for Connect {
|
||||
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
|
||||
write!(f, "{}:{}", self.host, self.port)
|
||||
}
|
||||
}
|
||||
|
||||
/// Tcp connector
|
||||
pub struct Connector {
|
||||
resolver: Resolver<Connect>,
|
||||
}
|
||||
|
||||
impl Default for Connector {
|
||||
fn default() -> Self {
|
||||
let (cfg, opts) = if let Ok((cfg, opts)) = read_system_conf() {
|
||||
(cfg, opts)
|
||||
@ -56,51 +132,37 @@ impl<T: HostAware> Default for Connector<T> {
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: HostAware> Connector<T> {
|
||||
impl Connector {
|
||||
/// Create new connector with resolver configuration
|
||||
pub fn new(cfg: ResolverConfig, opts: ResolverOpts) -> Self {
|
||||
Connector {
|
||||
resolver: Resolver::new(cfg, opts),
|
||||
}
|
||||
}
|
||||
|
||||
/// Create new connector with custom resolver
|
||||
pub fn with_resolver(
|
||||
resolver: Resolver<T>,
|
||||
) -> impl Service<
|
||||
Request = T,
|
||||
Response = (T, ConnectionInfo, TcpStream),
|
||||
Error = ConnectorError,
|
||||
> + Clone {
|
||||
resolver: Resolver<Connect>,
|
||||
) -> impl Service<Request = Connect, Response = (Connect, TcpStream), Error = ConnectorError>
|
||||
+ Clone {
|
||||
Connector { resolver }
|
||||
}
|
||||
|
||||
pub fn new_service<E>() -> impl NewService<
|
||||
Request = T,
|
||||
Response = (T, ConnectionInfo, TcpStream),
|
||||
Error = ConnectorError,
|
||||
InitError = E,
|
||||
> + Clone {
|
||||
|| -> FutureResult<Connector<T>, E> { ok(Connector::default()) }
|
||||
}
|
||||
|
||||
/// Create new default connector service
|
||||
pub fn new_service_with_config<E>(
|
||||
cfg: ResolverConfig, opts: ResolverOpts,
|
||||
cfg: ResolverConfig,
|
||||
opts: ResolverOpts,
|
||||
) -> impl NewService<
|
||||
Request = T,
|
||||
Response = (T, ConnectionInfo, TcpStream),
|
||||
Request = Connect,
|
||||
Response = (Connect, TcpStream),
|
||||
Error = ConnectorError,
|
||||
InitError = E,
|
||||
> + Clone {
|
||||
move || -> FutureResult<Connector<T>, E> { ok(Connector::new(cfg.clone(), opts)) }
|
||||
}
|
||||
|
||||
pub fn change_request<T2: HostAware>(&self) -> Connector<T2> {
|
||||
Connector {
|
||||
resolver: self.resolver.change_request(),
|
||||
}
|
||||
move || -> FutureResult<Connector, E> { ok(Connector::new(cfg.clone(), opts)) }
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> Clone for Connector<T> {
|
||||
impl Clone for Connector {
|
||||
fn clone(&self) -> Self {
|
||||
Connector {
|
||||
resolver: self.resolver.clone(),
|
||||
@ -108,11 +170,11 @@ impl<T> Clone for Connector<T> {
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: HostAware> Service for Connector<T> {
|
||||
type Request = T;
|
||||
type Response = (T, ConnectionInfo, TcpStream);
|
||||
impl Service for Connector {
|
||||
type Request = Connect;
|
||||
type Response = (Connect, TcpStream);
|
||||
type Error = ConnectorError;
|
||||
type Future = ConnectorFuture<T>;
|
||||
type Future = ConnectorFuture;
|
||||
|
||||
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
|
||||
Ok(Async::Ready(()))
|
||||
@ -127,25 +189,25 @@ impl<T: HostAware> Service for Connector<T> {
|
||||
}
|
||||
|
||||
#[doc(hidden)]
|
||||
pub struct ConnectorFuture<T: HostAware> {
|
||||
fut: ResolverFuture<T>,
|
||||
fut2: Option<TcpConnector<T>>,
|
||||
pub struct ConnectorFuture {
|
||||
fut: ResolverFuture<Connect>,
|
||||
fut2: Option<TcpConnectorResponse<Connect>>,
|
||||
}
|
||||
|
||||
impl<T: HostAware> Future for ConnectorFuture<T> {
|
||||
type Item = (T, ConnectionInfo, TcpStream);
|
||||
impl Future for ConnectorFuture {
|
||||
type Item = (Connect, TcpStream);
|
||||
type Error = ConnectorError;
|
||||
|
||||
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
|
||||
if let Some(ref mut fut) = self.fut2 {
|
||||
return fut.poll();
|
||||
return fut.poll().map_err(ConnectorError::from);
|
||||
}
|
||||
match self.fut.poll().map_err(ConnectorError::from)? {
|
||||
Async::Ready((req, host, addrs)) => {
|
||||
Async::Ready((req, addrs)) => {
|
||||
if addrs.is_empty() {
|
||||
Err(ConnectorError::NoRecords)
|
||||
} else {
|
||||
self.fut2 = Some(TcpConnector::new(req, host, addrs));
|
||||
self.fut2 = Some(TcpConnectorResponse::new(req, addrs));
|
||||
self.poll()
|
||||
}
|
||||
}
|
||||
@ -154,26 +216,101 @@ impl<T: HostAware> Future for ConnectorFuture<T> {
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct DefaultConnector<T: HostAware>(Connector<T>);
|
||||
/// Tcp stream connector service
|
||||
pub struct TcpConnector<T: RequestPort>(PhantomData<T>);
|
||||
|
||||
impl<T: HostAware> Default for DefaultConnector<T> {
|
||||
impl<T: RequestPort> Default for TcpConnector<T> {
|
||||
fn default() -> TcpConnector<T> {
|
||||
TcpConnector(PhantomData)
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: RequestPort> Service for TcpConnector<T> {
|
||||
type Request = (T, VecDeque<IpAddr>);
|
||||
type Response = (T, TcpStream);
|
||||
type Error = io::Error;
|
||||
type Future = TcpConnectorResponse<T>;
|
||||
|
||||
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
|
||||
Ok(Async::Ready(()))
|
||||
}
|
||||
|
||||
fn call(&mut self, (req, addrs): Self::Request) -> Self::Future {
|
||||
TcpConnectorResponse::new(req, addrs)
|
||||
}
|
||||
}
|
||||
|
||||
#[doc(hidden)]
|
||||
/// Tcp stream connector response future
|
||||
pub struct TcpConnectorResponse<T: RequestPort> {
|
||||
port: u16,
|
||||
req: Option<T>,
|
||||
addr: Option<SocketAddr>,
|
||||
addrs: VecDeque<IpAddr>,
|
||||
stream: Option<ConnectFuture>,
|
||||
}
|
||||
|
||||
impl<T: RequestPort> TcpConnectorResponse<T> {
|
||||
pub fn new(req: T, addrs: VecDeque<IpAddr>) -> TcpConnectorResponse<T> {
|
||||
TcpConnectorResponse {
|
||||
addrs,
|
||||
port: req.port(),
|
||||
req: Some(req),
|
||||
addr: None,
|
||||
stream: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: RequestPort> Future for TcpConnectorResponse<T> {
|
||||
type Item = (T, TcpStream);
|
||||
type Error = io::Error;
|
||||
|
||||
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
|
||||
// connect
|
||||
loop {
|
||||
if let Some(new) = self.stream.as_mut() {
|
||||
match new.poll() {
|
||||
Ok(Async::Ready(sock)) => {
|
||||
return Ok(Async::Ready((self.req.take().unwrap(), sock)))
|
||||
}
|
||||
Ok(Async::NotReady) => return Ok(Async::NotReady),
|
||||
Err(err) => {
|
||||
if self.addrs.is_empty() {
|
||||
return Err(err);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// try to connect
|
||||
let addr = SocketAddr::new(self.addrs.pop_front().unwrap(), self.port);
|
||||
self.stream = Some(TcpStream::connect(&addr));
|
||||
self.addr = Some(addr)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct DefaultConnector(Connector);
|
||||
|
||||
impl Default for DefaultConnector {
|
||||
fn default() -> Self {
|
||||
DefaultConnector(Connector::default())
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: HostAware> DefaultConnector<T> {
|
||||
impl DefaultConnector {
|
||||
pub fn new(cfg: ResolverConfig, opts: ResolverOpts) -> Self {
|
||||
DefaultConnector(Connector::new(cfg, opts))
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: HostAware> Service for DefaultConnector<T> {
|
||||
type Request = T;
|
||||
impl Service for DefaultConnector {
|
||||
type Request = Connect;
|
||||
type Response = TcpStream;
|
||||
type Error = ConnectorError;
|
||||
type Future = DefaultConnectorFuture<T>;
|
||||
type Future = DefaultConnectorFuture;
|
||||
|
||||
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
|
||||
self.0.poll_ready()
|
||||
@ -187,73 +324,15 @@ impl<T: HostAware> Service for DefaultConnector<T> {
|
||||
}
|
||||
|
||||
#[doc(hidden)]
|
||||
pub struct DefaultConnectorFuture<T: HostAware> {
|
||||
fut: ConnectorFuture<T>,
|
||||
pub struct DefaultConnectorFuture {
|
||||
fut: ConnectorFuture,
|
||||
}
|
||||
|
||||
impl<T: HostAware> Future for DefaultConnectorFuture<T> {
|
||||
impl Future for DefaultConnectorFuture {
|
||||
type Item = TcpStream;
|
||||
type Error = ConnectorError;
|
||||
|
||||
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
|
||||
Ok(Async::Ready(try_ready!(self.fut.poll()).2))
|
||||
}
|
||||
}
|
||||
|
||||
#[doc(hidden)]
|
||||
/// Tcp stream connector
|
||||
pub struct TcpConnector<T> {
|
||||
req: Option<T>,
|
||||
host: Option<String>,
|
||||
addr: Option<SocketAddr>,
|
||||
addrs: VecDeque<SocketAddr>,
|
||||
stream: Option<ConnectFuture>,
|
||||
}
|
||||
|
||||
impl<T> TcpConnector<T> {
|
||||
pub fn new(req: T, host: String, addrs: VecDeque<SocketAddr>) -> TcpConnector<T> {
|
||||
TcpConnector {
|
||||
addrs,
|
||||
req: Some(req),
|
||||
host: Some(host),
|
||||
addr: None,
|
||||
stream: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> Future for TcpConnector<T> {
|
||||
type Item = (T, ConnectionInfo, TcpStream);
|
||||
type Error = ConnectorError;
|
||||
|
||||
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
|
||||
// connect
|
||||
loop {
|
||||
if let Some(new) = self.stream.as_mut() {
|
||||
match new.poll() {
|
||||
Ok(Async::Ready(sock)) => {
|
||||
return Ok(Async::Ready((
|
||||
self.req.take().unwrap(),
|
||||
ConnectionInfo {
|
||||
host: self.host.take().unwrap(),
|
||||
addr: self.addr.take().unwrap(),
|
||||
},
|
||||
sock,
|
||||
)))
|
||||
}
|
||||
Ok(Async::NotReady) => return Ok(Async::NotReady),
|
||||
Err(err) => {
|
||||
if self.addrs.is_empty() {
|
||||
return Err(ConnectorError::IoError(err));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// try to connect
|
||||
let addr = self.addrs.pop_front().unwrap();
|
||||
self.stream = Some(TcpStream::connect(&addr));
|
||||
self.addr = Some(addr)
|
||||
}
|
||||
Ok(Async::Ready(try_ready!(self.fut.poll()).1))
|
||||
}
|
||||
}
|
||||
|
@ -6,13 +6,13 @@ use futures::{Async, Future, Poll};
|
||||
use tokio_timer::Delay;
|
||||
|
||||
use super::service::{NewService, Service};
|
||||
use super::timer::{LowResTimer, LowResTimerService};
|
||||
use super::time::{LowResTime, LowResTimeService};
|
||||
use super::Never;
|
||||
|
||||
pub struct KeepAlive<R, E, F> {
|
||||
f: F,
|
||||
ka: Duration,
|
||||
timer: LowResTimer,
|
||||
time: LowResTime,
|
||||
_t: PhantomData<(R, E)>,
|
||||
}
|
||||
|
||||
@ -20,11 +20,11 @@ impl<R, E, F> KeepAlive<R, E, F>
|
||||
where
|
||||
F: Fn() -> E + Clone,
|
||||
{
|
||||
pub fn new(ka: Duration, timer: LowResTimer, f: F) -> Self {
|
||||
pub fn new(ka: Duration, time: LowResTime, f: F) -> Self {
|
||||
KeepAlive {
|
||||
f,
|
||||
ka,
|
||||
timer,
|
||||
time,
|
||||
_t: PhantomData,
|
||||
}
|
||||
}
|
||||
@ -38,7 +38,7 @@ where
|
||||
KeepAlive {
|
||||
f: self.f.clone(),
|
||||
ka: self.ka,
|
||||
timer: self.timer.clone(),
|
||||
time: self.time.clone(),
|
||||
_t: PhantomData,
|
||||
}
|
||||
}
|
||||
@ -58,7 +58,7 @@ where
|
||||
fn new_service(&self) -> Self::Future {
|
||||
ok(KeepAliveService::new(
|
||||
self.ka,
|
||||
self.timer.timer(),
|
||||
self.time.timer(),
|
||||
self.f.clone(),
|
||||
))
|
||||
}
|
||||
@ -67,7 +67,7 @@ where
|
||||
pub struct KeepAliveService<R, E, F> {
|
||||
f: F,
|
||||
ka: Duration,
|
||||
timer: LowResTimerService,
|
||||
time: LowResTimeService,
|
||||
delay: Delay,
|
||||
expire: Instant,
|
||||
_t: PhantomData<(R, E)>,
|
||||
@ -77,14 +77,14 @@ impl<R, E, F> KeepAliveService<R, E, F>
|
||||
where
|
||||
F: Fn() -> E,
|
||||
{
|
||||
pub fn new(ka: Duration, mut timer: LowResTimerService, f: F) -> Self {
|
||||
let expire = timer.now() + ka;
|
||||
pub fn new(ka: Duration, time: LowResTimeService, f: F) -> Self {
|
||||
let expire = time.now() + ka;
|
||||
KeepAliveService {
|
||||
f,
|
||||
ka,
|
||||
timer,
|
||||
delay: Delay::new(expire),
|
||||
time,
|
||||
expire,
|
||||
delay: Delay::new(expire),
|
||||
_t: PhantomData,
|
||||
}
|
||||
}
|
||||
@ -102,11 +102,11 @@ where
|
||||
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
|
||||
match self.delay.poll() {
|
||||
Ok(Async::Ready(_)) => {
|
||||
let now = self.timer.now();
|
||||
let now = self.time.now();
|
||||
if self.expire <= now {
|
||||
Err((self.f)())
|
||||
} else {
|
||||
self.delay = Delay::new(self.expire);
|
||||
self.delay.reset(self.expire);
|
||||
let _ = self.delay.poll();
|
||||
Ok(Async::Ready(()))
|
||||
}
|
||||
@ -117,7 +117,7 @@ where
|
||||
}
|
||||
|
||||
fn call(&mut self, req: Self::Request) -> Self::Future {
|
||||
self.expire = self.timer.now() + self.ka;
|
||||
self.expire = self.time.now() + self.ka;
|
||||
ok(req)
|
||||
}
|
||||
}
|
||||
|
@ -69,7 +69,8 @@ pub mod server;
|
||||
pub mod service;
|
||||
pub mod ssl;
|
||||
pub mod stream;
|
||||
pub mod timer;
|
||||
pub mod time;
|
||||
pub mod timeout;
|
||||
|
||||
#[derive(Copy, Clone, Debug)]
|
||||
pub enum Never {}
|
||||
|
108
src/resolver.rs
108
src/resolver.rs
@ -1,29 +1,24 @@
|
||||
use std::collections::VecDeque;
|
||||
use std::marker::PhantomData;
|
||||
use std::net::SocketAddr;
|
||||
use std::net::IpAddr;
|
||||
|
||||
use futures::{Async, Future, Poll};
|
||||
|
||||
use tokio_current_thread::spawn;
|
||||
use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
|
||||
use trust_dns_resolver::error::ResolveError;
|
||||
pub use trust_dns_resolver::error::ResolveError;
|
||||
use trust_dns_resolver::lookup_ip::LookupIpFuture;
|
||||
use trust_dns_resolver::system_conf::read_system_conf;
|
||||
use trust_dns_resolver::{AsyncResolver, Background};
|
||||
|
||||
use super::service::Service;
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum ResolverError {
|
||||
Resolve(ResolveError),
|
||||
InvalidInput,
|
||||
}
|
||||
|
||||
pub trait HostAware {
|
||||
/// Host name of the request
|
||||
pub trait RequestHost {
|
||||
fn host(&self) -> &str;
|
||||
}
|
||||
|
||||
impl HostAware for String {
|
||||
impl RequestHost for String {
|
||||
fn host(&self) -> &str {
|
||||
self.as_ref()
|
||||
}
|
||||
@ -34,7 +29,7 @@ pub struct Resolver<T = String> {
|
||||
req: PhantomData<T>,
|
||||
}
|
||||
|
||||
impl<T: HostAware> Default for Resolver<T> {
|
||||
impl<T: RequestHost> Default for Resolver<T> {
|
||||
fn default() -> Self {
|
||||
let (cfg, opts) = if let Ok((cfg, opts)) = read_system_conf() {
|
||||
(cfg, opts)
|
||||
@ -46,7 +41,8 @@ impl<T: HostAware> Default for Resolver<T> {
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: HostAware> Resolver<T> {
|
||||
impl<T: RequestHost> Resolver<T> {
|
||||
/// Create new resolver instance with custom configuration and options.
|
||||
pub fn new(cfg: ResolverConfig, opts: ResolverOpts) -> Self {
|
||||
let (resolver, bg) = AsyncResolver::new(cfg, opts);
|
||||
spawn(bg);
|
||||
@ -56,7 +52,8 @@ impl<T: HostAware> Resolver<T> {
|
||||
}
|
||||
}
|
||||
|
||||
pub fn change_request<T2: HostAware>(&self) -> Resolver<T2> {
|
||||
/// Change type of resolver request.
|
||||
pub fn into_request<T2: RequestHost>(&self) -> Resolver<T2> {
|
||||
Resolver {
|
||||
resolver: self.resolver.clone(),
|
||||
req: PhantomData,
|
||||
@ -73,10 +70,10 @@ impl<T> Clone for Resolver<T> {
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: HostAware> Service for Resolver<T> {
|
||||
impl<T: RequestHost> Service for Resolver<T> {
|
||||
type Request = T;
|
||||
type Response = (T, String, VecDeque<SocketAddr>);
|
||||
type Error = ResolverError;
|
||||
type Response = (T, VecDeque<IpAddr>);
|
||||
type Error = ResolveError;
|
||||
type Future = ResolverFuture<T>;
|
||||
|
||||
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
|
||||
@ -84,7 +81,7 @@ impl<T: HostAware> Service for Resolver<T> {
|
||||
}
|
||||
|
||||
fn call(&mut self, req: Self::Request) -> Self::Future {
|
||||
ResolverFuture::new(req, 0, &self.resolver)
|
||||
ResolverFuture::new(req, &self.resolver)
|
||||
}
|
||||
}
|
||||
|
||||
@ -92,78 +89,37 @@ impl<T: HostAware> Service for Resolver<T> {
|
||||
/// Resolver future
|
||||
pub struct ResolverFuture<T> {
|
||||
req: Option<T>,
|
||||
port: u16,
|
||||
lookup: Option<Background<LookupIpFuture>>,
|
||||
addrs: Option<VecDeque<SocketAddr>>,
|
||||
error: Option<ResolverError>,
|
||||
host: Option<String>,
|
||||
addrs: Option<VecDeque<IpAddr>>,
|
||||
}
|
||||
|
||||
impl<T: HostAware> ResolverFuture<T> {
|
||||
pub fn new(addr: T, port: u16, resolver: &AsyncResolver) -> Self {
|
||||
impl<T: RequestHost> ResolverFuture<T> {
|
||||
pub fn new(addr: T, resolver: &AsyncResolver) -> Self {
|
||||
// we need to do dns resolution
|
||||
match ResolverFuture::<T>::parse(addr.host(), port) {
|
||||
Ok((host, port)) => {
|
||||
let lookup = Some(resolver.lookup_ip(host.as_str()));
|
||||
ResolverFuture {
|
||||
port,
|
||||
lookup,
|
||||
req: Some(addr),
|
||||
host: Some(host.to_owned()),
|
||||
addrs: None,
|
||||
error: None,
|
||||
}
|
||||
}
|
||||
Err(err) => ResolverFuture {
|
||||
port,
|
||||
req: None,
|
||||
host: None,
|
||||
lookup: None,
|
||||
addrs: None,
|
||||
error: Some(err),
|
||||
},
|
||||
let lookup = Some(resolver.lookup_ip(addr.host()));
|
||||
ResolverFuture {
|
||||
lookup,
|
||||
req: Some(addr),
|
||||
addrs: None,
|
||||
}
|
||||
}
|
||||
|
||||
fn parse(addr: &str, port: u16) -> Result<(String, u16), ResolverError> {
|
||||
// split the string by ':' and convert the second part to u16
|
||||
let mut parts_iter = addr.splitn(2, ':');
|
||||
let host = parts_iter.next().ok_or(ResolverError::InvalidInput)?;
|
||||
let port_str = parts_iter.next().unwrap_or("");
|
||||
let port: u16 = port_str.parse().unwrap_or(port);
|
||||
|
||||
Ok((host.to_owned(), port))
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: HostAware> Future for ResolverFuture<T> {
|
||||
type Item = (T, String, VecDeque<SocketAddr>);
|
||||
type Error = ResolverError;
|
||||
impl<T: RequestHost> Future for ResolverFuture<T> {
|
||||
type Item = (T, VecDeque<IpAddr>);
|
||||
type Error = ResolveError;
|
||||
|
||||
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
|
||||
if let Some(err) = self.error.take() {
|
||||
Err(err)
|
||||
} else if let Some(addrs) = self.addrs.take() {
|
||||
Ok(Async::Ready((
|
||||
self.req.take().unwrap(),
|
||||
self.host.take().unwrap(),
|
||||
addrs,
|
||||
)))
|
||||
if let Some(addrs) = self.addrs.take() {
|
||||
Ok(Async::Ready((self.req.take().unwrap(), addrs)))
|
||||
} else {
|
||||
match self.lookup.as_mut().unwrap().poll() {
|
||||
Ok(Async::NotReady) => Ok(Async::NotReady),
|
||||
Ok(Async::Ready(ips)) => {
|
||||
let addrs: VecDeque<_> = ips
|
||||
.iter()
|
||||
.map(|ip| SocketAddr::new(ip, self.port))
|
||||
.collect();
|
||||
Ok(Async::Ready((
|
||||
self.req.take().unwrap(),
|
||||
self.host.take().unwrap(),
|
||||
addrs,
|
||||
)))
|
||||
}
|
||||
Err(err) => Err(ResolverError::Resolve(err)),
|
||||
Ok(Async::Ready(ips)) => Ok(Async::Ready((
|
||||
self.req.take().unwrap(),
|
||||
ips.iter().collect(),
|
||||
))),
|
||||
Err(err) => Err(err),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -23,7 +23,6 @@ pub(crate) enum Command {
|
||||
struct ServerSocketInfo {
|
||||
addr: net::SocketAddr,
|
||||
token: Token,
|
||||
handler: Token,
|
||||
sock: mio::net::TcpListener,
|
||||
timeout: Option<Instant>,
|
||||
}
|
||||
@ -87,7 +86,9 @@ impl AcceptLoop {
|
||||
}
|
||||
|
||||
pub(crate) fn start(
|
||||
&mut self, socks: Vec<(Token, net::TcpListener)>, workers: Vec<WorkerClient>,
|
||||
&mut self,
|
||||
socks: Vec<(Token, net::TcpListener)>,
|
||||
workers: Vec<WorkerClient>,
|
||||
) -> mpsc::UnboundedReceiver<ServerCommand> {
|
||||
let (tx, rx) = self.srv.take().expect("Can not re-use AcceptInfo");
|
||||
|
||||
@ -135,9 +136,12 @@ fn connection_error(e: &io::Error) -> bool {
|
||||
impl Accept {
|
||||
#![cfg_attr(feature = "cargo-clippy", allow(too_many_arguments))]
|
||||
pub(crate) fn start(
|
||||
rx: sync_mpsc::Receiver<Command>, cmd_reg: mio::Registration,
|
||||
notify_reg: mio::Registration, socks: Vec<(Token, net::TcpListener)>,
|
||||
srv: mpsc::UnboundedSender<ServerCommand>, workers: Vec<WorkerClient>,
|
||||
rx: sync_mpsc::Receiver<Command>,
|
||||
cmd_reg: mio::Registration,
|
||||
notify_reg: mio::Registration,
|
||||
socks: Vec<(Token, net::TcpListener)>,
|
||||
srv: mpsc::UnboundedSender<ServerCommand>,
|
||||
workers: Vec<WorkerClient>,
|
||||
) {
|
||||
let sys = System::current();
|
||||
|
||||
@ -173,8 +177,10 @@ impl Accept {
|
||||
}
|
||||
|
||||
fn new(
|
||||
rx: sync_mpsc::Receiver<Command>, socks: Vec<(Token, net::TcpListener)>,
|
||||
workers: Vec<WorkerClient>, srv: mpsc::UnboundedSender<ServerCommand>,
|
||||
rx: sync_mpsc::Receiver<Command>,
|
||||
socks: Vec<(Token, net::TcpListener)>,
|
||||
workers: Vec<WorkerClient>,
|
||||
srv: mpsc::UnboundedSender<ServerCommand>,
|
||||
) -> Accept {
|
||||
// Create a poll instance
|
||||
let poll = match mio::Poll::new() {
|
||||
@ -184,7 +190,7 @@ impl Accept {
|
||||
|
||||
// Start accept
|
||||
let mut sockets = Slab::new();
|
||||
for (idx, (hnd_token, lst)) in socks.into_iter().enumerate() {
|
||||
for (hnd_token, lst) in socks.into_iter() {
|
||||
let addr = lst.local_addr().unwrap();
|
||||
let server = mio::net::TcpListener::from_std(lst)
|
||||
.expect("Can not create mio::net::TcpListener");
|
||||
@ -205,7 +211,6 @@ impl Accept {
|
||||
entry.insert(ServerSocketInfo {
|
||||
addr,
|
||||
token: hnd_token,
|
||||
handler: Token(idx),
|
||||
sock: server,
|
||||
timeout: None,
|
||||
});
|
||||
@ -243,9 +248,11 @@ impl Accept {
|
||||
for event in events.iter() {
|
||||
let token = event.token();
|
||||
match token {
|
||||
CMD => if !self.process_cmd() {
|
||||
return;
|
||||
},
|
||||
CMD => {
|
||||
if !self.process_cmd() {
|
||||
return;
|
||||
}
|
||||
}
|
||||
TIMER => self.process_timer(),
|
||||
NOTIFY => self.backpressure(false),
|
||||
_ => {
|
||||
@ -427,7 +434,6 @@ impl Accept {
|
||||
Ok((io, addr)) => Conn {
|
||||
io,
|
||||
token: info.token,
|
||||
handler: info.handler,
|
||||
peer: Some(addr),
|
||||
},
|
||||
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => return,
|
||||
|
213
src/server/config.rs
Normal file
213
src/server/config.rs
Normal file
@ -0,0 +1,213 @@
|
||||
use std::collections::HashMap;
|
||||
use std::{fmt, io, net};
|
||||
|
||||
use futures::future::{join_all, Future};
|
||||
use tokio_tcp::TcpStream;
|
||||
|
||||
use counter::CounterGuard;
|
||||
use service::{IntoNewService, NewService};
|
||||
|
||||
use super::server::bind_addr;
|
||||
use super::services::{
|
||||
BoxedServerService, InternalServiceFactory, ServerMessage, StreamService,
|
||||
};
|
||||
use super::Token;
|
||||
|
||||
pub struct ServiceConfig {
|
||||
pub(super) services: Vec<(String, net::TcpListener)>,
|
||||
pub(super) rt: Box<ServiceRuntimeConfiguration>,
|
||||
}
|
||||
|
||||
impl ServiceConfig {
|
||||
pub(super) fn new() -> ServiceConfig {
|
||||
ServiceConfig {
|
||||
services: Vec::new(),
|
||||
rt: Box::new(not_configured),
|
||||
}
|
||||
}
|
||||
|
||||
/// Add new service to server
|
||||
pub fn bind<U, N: AsRef<str>>(&mut self, name: N, addr: U) -> io::Result<&mut Self>
|
||||
where
|
||||
U: net::ToSocketAddrs,
|
||||
{
|
||||
let sockets = bind_addr(addr)?;
|
||||
|
||||
for lst in sockets {
|
||||
self.listen(name.as_ref(), lst);
|
||||
}
|
||||
|
||||
Ok(self)
|
||||
}
|
||||
|
||||
/// Add new service to server
|
||||
pub fn listen<N: AsRef<str>>(&mut self, name: N, lst: net::TcpListener) -> &mut Self {
|
||||
self.services.push((name.as_ref().to_string(), lst));
|
||||
self
|
||||
}
|
||||
|
||||
/// Register service configuration function
|
||||
pub fn rt<F>(&mut self, f: F) -> io::Result<()>
|
||||
where
|
||||
F: Fn(&mut ServiceRuntime) + Send + Clone + 'static,
|
||||
{
|
||||
self.rt = Box::new(f);
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) struct ConfiguredService {
|
||||
rt: Box<ServiceRuntimeConfiguration>,
|
||||
names: HashMap<Token, String>,
|
||||
services: HashMap<String, Token>,
|
||||
}
|
||||
|
||||
impl ConfiguredService {
|
||||
pub(super) fn new(rt: Box<ServiceRuntimeConfiguration>) -> Self {
|
||||
ConfiguredService {
|
||||
rt,
|
||||
names: HashMap::new(),
|
||||
services: HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn stream(&mut self, token: Token, name: String) {
|
||||
self.names.insert(token, name.clone());
|
||||
self.services.insert(name, token);
|
||||
}
|
||||
}
|
||||
|
||||
impl InternalServiceFactory for ConfiguredService {
|
||||
fn name(&self, token: Token) -> &str {
|
||||
&self.names[&token]
|
||||
}
|
||||
|
||||
fn clone_factory(&self) -> Box<InternalServiceFactory> {
|
||||
Box::new(Self {
|
||||
rt: self.rt.clone(),
|
||||
names: self.names.clone(),
|
||||
services: self.services.clone(),
|
||||
})
|
||||
}
|
||||
|
||||
fn create(&self) -> Box<Future<Item = Vec<(Token, BoxedServerService)>, Error = ()>> {
|
||||
// configure services
|
||||
let mut rt = ServiceRuntime::new(self.services.clone());
|
||||
self.rt.configure(&mut rt);
|
||||
rt.validate();
|
||||
|
||||
// construct services
|
||||
let mut fut = Vec::new();
|
||||
for (token, ns) in rt.services {
|
||||
fut.push(ns.new_service().map(move |service| (token, service)));
|
||||
}
|
||||
|
||||
Box::new(join_all(fut).map_err(|e| {
|
||||
error!("Can not construct service: {:?}", e);
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) trait ServiceRuntimeConfiguration: Send {
|
||||
fn clone(&self) -> Box<ServiceRuntimeConfiguration>;
|
||||
|
||||
fn configure(&self, &mut ServiceRuntime);
|
||||
}
|
||||
|
||||
impl<F> ServiceRuntimeConfiguration for F
|
||||
where
|
||||
F: Fn(&mut ServiceRuntime) + Send + Clone + 'static,
|
||||
{
|
||||
fn clone(&self) -> Box<ServiceRuntimeConfiguration> {
|
||||
Box::new(self.clone())
|
||||
}
|
||||
|
||||
fn configure(&self, rt: &mut ServiceRuntime) {
|
||||
(self)(rt)
|
||||
}
|
||||
}
|
||||
|
||||
fn not_configured(_: &mut ServiceRuntime) {
|
||||
error!("Service is not configured");
|
||||
}
|
||||
|
||||
pub struct ServiceRuntime {
|
||||
names: HashMap<String, Token>,
|
||||
services: HashMap<Token, BoxedNewService>,
|
||||
}
|
||||
|
||||
impl ServiceRuntime {
|
||||
fn new(names: HashMap<String, Token>) -> Self {
|
||||
ServiceRuntime {
|
||||
names,
|
||||
services: HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
fn validate(&self) {
|
||||
for (name, token) in &self.names {
|
||||
if !self.services.contains_key(&token) {
|
||||
error!("Service {:?} is not configured", name);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn service<T, F>(&mut self, name: &str, service: F)
|
||||
where
|
||||
F: IntoNewService<T>,
|
||||
T: NewService<Request = TcpStream, Response = ()> + 'static,
|
||||
T::Future: 'static,
|
||||
T::Service: 'static,
|
||||
T::InitError: fmt::Debug,
|
||||
{
|
||||
// let name = name.to_owned();
|
||||
if let Some(token) = self.names.get(name) {
|
||||
self.services.insert(
|
||||
token.clone(),
|
||||
Box::new(ServiceFactory {
|
||||
inner: service.into_new_service(),
|
||||
}),
|
||||
);
|
||||
} else {
|
||||
panic!("Unknown service: {:?}", name);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type BoxedNewService = Box<
|
||||
NewService<
|
||||
Request = (Option<CounterGuard>, ServerMessage),
|
||||
Response = (),
|
||||
Error = (),
|
||||
InitError = (),
|
||||
Service = BoxedServerService,
|
||||
Future = Box<Future<Item = BoxedServerService, Error = ()>>,
|
||||
>,
|
||||
>;
|
||||
|
||||
struct ServiceFactory<T> {
|
||||
inner: T,
|
||||
}
|
||||
|
||||
impl<T> NewService for ServiceFactory<T>
|
||||
where
|
||||
T: NewService<Request = TcpStream, Response = ()>,
|
||||
T::Future: 'static,
|
||||
T::Service: 'static,
|
||||
T::Error: 'static,
|
||||
T::InitError: fmt::Debug + 'static,
|
||||
{
|
||||
type Request = (Option<CounterGuard>, ServerMessage);
|
||||
type Response = ();
|
||||
type Error = ();
|
||||
type InitError = ();
|
||||
type Service = BoxedServerService;
|
||||
type Future = Box<Future<Item = BoxedServerService, Error = ()>>;
|
||||
|
||||
fn new_service(&self) -> Self::Future {
|
||||
Box::new(self.inner.new_service().map_err(|_| ()).map(|s| {
|
||||
let service: BoxedServerService = Box::new(StreamService::new(s));
|
||||
service
|
||||
}))
|
||||
}
|
||||
}
|
@ -3,10 +3,12 @@
|
||||
use actix::Message;
|
||||
|
||||
mod accept;
|
||||
mod config;
|
||||
mod server;
|
||||
mod services;
|
||||
mod worker;
|
||||
|
||||
pub use self::config::{ServiceConfig, ServiceRuntime};
|
||||
pub use self::server::Server;
|
||||
pub use self::services::{ServerMessage, ServiceFactory, StreamServiceFactory};
|
||||
|
||||
@ -34,5 +36,13 @@ impl Message for StopServer {
|
||||
}
|
||||
|
||||
/// Socket id token
|
||||
#[derive(Clone, Copy, Debug)]
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
|
||||
pub(crate) struct Token(usize);
|
||||
|
||||
impl Token {
|
||||
pub(crate) fn next(&mut self) -> Token {
|
||||
let token = Token(self.0 + 1);
|
||||
self.0 += 1;
|
||||
token
|
||||
}
|
||||
}
|
||||
|
@ -12,6 +12,7 @@ use actix::{
|
||||
};
|
||||
|
||||
use super::accept::{AcceptLoop, AcceptNotify, Command};
|
||||
use super::config::{ConfiguredService, ServiceConfig};
|
||||
use super::services::{InternalServiceFactory, StreamNewService, StreamServiceFactory};
|
||||
use super::services::{ServiceFactory, ServiceNewService};
|
||||
use super::worker::{self, Worker, WorkerAvailability, WorkerClient};
|
||||
@ -24,6 +25,7 @@ pub(crate) enum ServerCommand {
|
||||
/// Server
|
||||
pub struct Server {
|
||||
threads: usize,
|
||||
token: Token,
|
||||
workers: Vec<(usize, WorkerClient)>,
|
||||
services: Vec<Box<InternalServiceFactory>>,
|
||||
sockets: Vec<(Token, net::TcpListener)>,
|
||||
@ -45,6 +47,7 @@ impl Server {
|
||||
pub fn new() -> Server {
|
||||
Server {
|
||||
threads: num_cpus::get(),
|
||||
token: Token(0),
|
||||
workers: Vec::new(),
|
||||
services: Vec::new(),
|
||||
sockets: Vec::new(),
|
||||
@ -113,12 +116,24 @@ impl Server {
|
||||
/// process
|
||||
///
|
||||
/// This function is useful for moving parts of configuration to a
|
||||
/// different module or event library.
|
||||
pub fn configure<F>(self, cfg: F) -> Server
|
||||
/// different module or even library.
|
||||
pub fn configure<F>(mut self, f: F) -> io::Result<Server>
|
||||
where
|
||||
F: Fn(Server) -> Server,
|
||||
F: Fn(&mut ServiceConfig) -> io::Result<()>,
|
||||
{
|
||||
cfg(self)
|
||||
let mut cfg = ServiceConfig::new();
|
||||
|
||||
f(&mut cfg)?;
|
||||
|
||||
let mut srv = ConfiguredService::new(cfg.rt);
|
||||
for (name, lst) in cfg.services {
|
||||
let token = self.token.next();
|
||||
srv.stream(token, name);
|
||||
self.sockets.push((token, lst));
|
||||
}
|
||||
self.services.push(Box::new(srv));
|
||||
|
||||
Ok(self)
|
||||
}
|
||||
|
||||
/// Add new service to server
|
||||
@ -129,36 +144,53 @@ impl Server {
|
||||
{
|
||||
let sockets = bind_addr(addr)?;
|
||||
|
||||
let token = self.token.next();
|
||||
self.services.push(StreamNewService::create(
|
||||
name.as_ref().to_string(),
|
||||
token,
|
||||
factory,
|
||||
));
|
||||
|
||||
for lst in sockets {
|
||||
self = self.listen(name.as_ref(), lst, factory.clone())
|
||||
self.sockets.push((token, lst));
|
||||
}
|
||||
Ok(self)
|
||||
}
|
||||
|
||||
/// Add new service to server
|
||||
pub fn listen<F, N: AsRef<str>>(
|
||||
mut self, name: N, lst: net::TcpListener, factory: F,
|
||||
mut self,
|
||||
name: N,
|
||||
lst: net::TcpListener,
|
||||
factory: F,
|
||||
) -> Self
|
||||
where
|
||||
F: StreamServiceFactory,
|
||||
{
|
||||
let token = Token(self.services.len());
|
||||
self.services
|
||||
.push(StreamNewService::create(name.as_ref().to_string(), factory));
|
||||
let token = self.token.next();
|
||||
self.services.push(StreamNewService::create(
|
||||
name.as_ref().to_string(),
|
||||
token,
|
||||
factory,
|
||||
));
|
||||
self.sockets.push((token, lst));
|
||||
self
|
||||
}
|
||||
|
||||
/// Add new service to server
|
||||
pub fn listen2<F, N: AsRef<str>>(
|
||||
mut self, name: N, lst: net::TcpListener, factory: F,
|
||||
mut self,
|
||||
name: N,
|
||||
lst: net::TcpListener,
|
||||
factory: F,
|
||||
) -> Self
|
||||
where
|
||||
F: ServiceFactory,
|
||||
{
|
||||
let token = Token(self.services.len());
|
||||
let token = self.token.next();
|
||||
self.services.push(ServiceNewService::create(
|
||||
name.as_ref().to_string(),
|
||||
token,
|
||||
factory,
|
||||
));
|
||||
self.sockets.push((token, lst));
|
||||
@ -243,15 +275,16 @@ impl Server {
|
||||
}
|
||||
|
||||
fn start_worker(&self, idx: usize, notify: AcceptNotify) -> WorkerClient {
|
||||
let (tx, rx) = unbounded();
|
||||
let (tx1, rx1) = unbounded();
|
||||
let (tx2, rx2) = unbounded();
|
||||
let timeout = self.shutdown_timeout;
|
||||
let avail = WorkerAvailability::new(notify);
|
||||
let worker = WorkerClient::new(idx, tx, avail.clone());
|
||||
let worker = WorkerClient::new(idx, tx1, tx2, avail.clone());
|
||||
let services: Vec<Box<InternalServiceFactory>> =
|
||||
self.services.iter().map(|v| v.clone_factory()).collect();
|
||||
|
||||
Arbiter::new(format!("actix-net-worker-{}", idx)).do_send(Execute::new(move || {
|
||||
Worker::start(rx, services, avail, timeout.clone());
|
||||
Worker::start(rx1, rx2, services, avail, timeout.clone());
|
||||
Ok::<_, ()>(())
|
||||
}));
|
||||
|
||||
@ -395,7 +428,7 @@ impl StreamHandler<ServerCommand, ()> for Server {
|
||||
}
|
||||
}
|
||||
|
||||
fn bind_addr<S: net::ToSocketAddrs>(addr: S) -> io::Result<Vec<net::TcpListener>> {
|
||||
pub(super) fn bind_addr<S: net::ToSocketAddrs>(addr: S) -> io::Result<Vec<net::TcpListener>> {
|
||||
let mut err = None;
|
||||
let mut succ = false;
|
||||
let mut sockets = Vec::new();
|
||||
|
@ -7,6 +7,7 @@ use tokio_current_thread::spawn;
|
||||
use tokio_reactor::Handle;
|
||||
use tokio_tcp::TcpStream;
|
||||
|
||||
use super::Token;
|
||||
use counter::CounterGuard;
|
||||
use service::{NewService, Service};
|
||||
|
||||
@ -33,11 +34,11 @@ pub trait ServiceFactory: Send + Clone + 'static {
|
||||
}
|
||||
|
||||
pub(crate) trait InternalServiceFactory: Send {
|
||||
fn name(&self) -> &str;
|
||||
fn name(&self, token: Token) -> &str;
|
||||
|
||||
fn clone_factory(&self) -> Box<InternalServiceFactory>;
|
||||
|
||||
fn create(&self) -> Box<Future<Item = BoxedServerService, Error = ()>>;
|
||||
fn create(&self) -> Box<Future<Item = Vec<(Token, BoxedServerService)>, Error = ()>>;
|
||||
}
|
||||
|
||||
pub(crate) type BoxedServerService = Box<
|
||||
@ -54,7 +55,7 @@ pub(crate) struct StreamService<T> {
|
||||
}
|
||||
|
||||
impl<T> StreamService<T> {
|
||||
fn new(service: T) -> Self {
|
||||
pub(crate) fn new(service: T) -> Self {
|
||||
StreamService { service }
|
||||
}
|
||||
}
|
||||
@ -133,14 +134,15 @@ where
|
||||
pub(crate) struct ServiceNewService<F: ServiceFactory> {
|
||||
name: String,
|
||||
inner: F,
|
||||
token: Token,
|
||||
}
|
||||
|
||||
impl<F> ServiceNewService<F>
|
||||
where
|
||||
F: ServiceFactory,
|
||||
{
|
||||
pub(crate) fn create(name: String, inner: F) -> Box<InternalServiceFactory> {
|
||||
Box::new(Self { name, inner })
|
||||
pub(crate) fn create(name: String, token: Token, inner: F) -> Box<InternalServiceFactory> {
|
||||
Box::new(Self { name, inner, token })
|
||||
}
|
||||
}
|
||||
|
||||
@ -148,7 +150,7 @@ impl<F> InternalServiceFactory for ServiceNewService<F>
|
||||
where
|
||||
F: ServiceFactory,
|
||||
{
|
||||
fn name(&self) -> &str {
|
||||
fn name(&self, _: Token) -> &str {
|
||||
&self.name
|
||||
}
|
||||
|
||||
@ -156,10 +158,12 @@ where
|
||||
Box::new(Self {
|
||||
name: self.name.clone(),
|
||||
inner: self.inner.clone(),
|
||||
token: self.token,
|
||||
})
|
||||
}
|
||||
|
||||
fn create(&self) -> Box<Future<Item = BoxedServerService, Error = ()>> {
|
||||
fn create(&self) -> Box<Future<Item = Vec<(Token, BoxedServerService)>, Error = ()>> {
|
||||
let token = self.token;
|
||||
Box::new(
|
||||
self.inner
|
||||
.create()
|
||||
@ -167,7 +171,7 @@ where
|
||||
.map_err(|_| ())
|
||||
.map(move |inner| {
|
||||
let service: BoxedServerService = Box::new(ServerService::new(inner));
|
||||
service
|
||||
vec![(token, service)]
|
||||
}),
|
||||
)
|
||||
}
|
||||
@ -176,14 +180,15 @@ where
|
||||
pub(crate) struct StreamNewService<F: StreamServiceFactory> {
|
||||
name: String,
|
||||
inner: F,
|
||||
token: Token,
|
||||
}
|
||||
|
||||
impl<F> StreamNewService<F>
|
||||
where
|
||||
F: StreamServiceFactory,
|
||||
{
|
||||
pub(crate) fn create(name: String, inner: F) -> Box<InternalServiceFactory> {
|
||||
Box::new(Self { name, inner })
|
||||
pub(crate) fn create(name: String, token: Token, inner: F) -> Box<InternalServiceFactory> {
|
||||
Box::new(Self { name, token, inner })
|
||||
}
|
||||
}
|
||||
|
||||
@ -191,7 +196,7 @@ impl<F> InternalServiceFactory for StreamNewService<F>
|
||||
where
|
||||
F: StreamServiceFactory,
|
||||
{
|
||||
fn name(&self) -> &str {
|
||||
fn name(&self, _: Token) -> &str {
|
||||
&self.name
|
||||
}
|
||||
|
||||
@ -199,10 +204,12 @@ where
|
||||
Box::new(Self {
|
||||
name: self.name.clone(),
|
||||
inner: self.inner.clone(),
|
||||
token: self.token,
|
||||
})
|
||||
}
|
||||
|
||||
fn create(&self) -> Box<Future<Item = BoxedServerService, Error = ()>> {
|
||||
fn create(&self) -> Box<Future<Item = Vec<(Token, BoxedServerService)>, Error = ()>> {
|
||||
let token = self.token;
|
||||
Box::new(
|
||||
self.inner
|
||||
.create()
|
||||
@ -210,22 +217,22 @@ where
|
||||
.map_err(|_| ())
|
||||
.map(move |inner| {
|
||||
let service: BoxedServerService = Box::new(StreamService::new(inner));
|
||||
service
|
||||
vec![(token, service)]
|
||||
}),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
impl InternalServiceFactory for Box<InternalServiceFactory> {
|
||||
fn name(&self) -> &str {
|
||||
self.as_ref().name()
|
||||
fn name(&self, token: Token) -> &str {
|
||||
self.as_ref().name(token)
|
||||
}
|
||||
|
||||
fn clone_factory(&self) -> Box<InternalServiceFactory> {
|
||||
self.as_ref().clone_factory()
|
||||
}
|
||||
|
||||
fn create(&self) -> Box<Future<Item = BoxedServerService, Error = ()>> {
|
||||
fn create(&self) -> Box<Future<Item = Vec<(Token, BoxedServerService)>, Error = ()>> {
|
||||
self.as_ref().create()
|
||||
}
|
||||
}
|
||||
|
@ -16,17 +16,18 @@ use super::services::{BoxedServerService, InternalServiceFactory, ServerMessage}
|
||||
use super::Token;
|
||||
use counter::Counter;
|
||||
|
||||
pub(crate) enum WorkerCommand {
|
||||
Message(Conn),
|
||||
/// Stop worker message. Returns `true` on successful shutdown
|
||||
/// and `false` if some connections still alive.
|
||||
Stop(bool, oneshot::Sender<bool>),
|
||||
pub(crate) struct WorkerCommand(Conn);
|
||||
|
||||
/// Stop worker message. Returns `true` on successful shutdown
|
||||
/// and `false` if some connections still alive.
|
||||
pub(crate) struct StopCommand {
|
||||
graceful: bool,
|
||||
result: oneshot::Sender<bool>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Message)]
|
||||
pub(crate) struct Conn {
|
||||
pub io: net::TcpStream,
|
||||
pub handler: Token,
|
||||
pub token: Token,
|
||||
pub peer: Option<net::SocketAddr>,
|
||||
}
|
||||
@ -55,24 +56,30 @@ thread_local! {
|
||||
#[derive(Clone)]
|
||||
pub(crate) struct WorkerClient {
|
||||
pub idx: usize,
|
||||
tx: UnboundedSender<WorkerCommand>,
|
||||
tx1: UnboundedSender<WorkerCommand>,
|
||||
tx2: UnboundedSender<StopCommand>,
|
||||
avail: WorkerAvailability,
|
||||
}
|
||||
|
||||
impl WorkerClient {
|
||||
pub fn new(
|
||||
idx: usize, tx: UnboundedSender<WorkerCommand>, avail: WorkerAvailability,
|
||||
idx: usize,
|
||||
tx1: UnboundedSender<WorkerCommand>,
|
||||
tx2: UnboundedSender<StopCommand>,
|
||||
avail: WorkerAvailability,
|
||||
) -> Self {
|
||||
WorkerClient { idx, tx, avail }
|
||||
WorkerClient {
|
||||
idx,
|
||||
tx1,
|
||||
tx2,
|
||||
avail,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn send(&self, msg: Conn) -> Result<(), Conn> {
|
||||
self.tx
|
||||
.unbounded_send(WorkerCommand::Message(msg))
|
||||
.map_err(|e| match e.into_inner() {
|
||||
WorkerCommand::Message(msg) => msg,
|
||||
_ => panic!(),
|
||||
})
|
||||
self.tx1
|
||||
.unbounded_send(WorkerCommand(msg))
|
||||
.map_err(|msg| msg.into_inner().0)
|
||||
}
|
||||
|
||||
pub fn available(&self) -> bool {
|
||||
@ -80,8 +87,8 @@ impl WorkerClient {
|
||||
}
|
||||
|
||||
pub fn stop(&self, graceful: bool) -> oneshot::Receiver<bool> {
|
||||
let (tx, rx) = oneshot::channel();
|
||||
let _ = self.tx.unbounded_send(WorkerCommand::Stop(graceful, tx));
|
||||
let (result, rx) = oneshot::channel();
|
||||
let _ = self.tx2.unbounded_send(StopCommand { graceful, result });
|
||||
rx
|
||||
}
|
||||
}
|
||||
@ -118,7 +125,8 @@ impl WorkerAvailability {
|
||||
/// processing.
|
||||
pub(crate) struct Worker {
|
||||
rx: UnboundedReceiver<WorkerCommand>,
|
||||
services: Vec<BoxedServerService>,
|
||||
rx2: UnboundedReceiver<StopCommand>,
|
||||
services: Vec<Option<(usize, BoxedServerService)>>,
|
||||
availability: WorkerAvailability,
|
||||
conns: Counter,
|
||||
factories: Vec<Box<InternalServiceFactory>>,
|
||||
@ -128,12 +136,16 @@ pub(crate) struct Worker {
|
||||
|
||||
impl Worker {
|
||||
pub(crate) fn start(
|
||||
rx: UnboundedReceiver<WorkerCommand>, factories: Vec<Box<InternalServiceFactory>>,
|
||||
availability: WorkerAvailability, shutdown_timeout: time::Duration,
|
||||
rx: UnboundedReceiver<WorkerCommand>,
|
||||
rx2: UnboundedReceiver<StopCommand>,
|
||||
factories: Vec<Box<InternalServiceFactory>>,
|
||||
availability: WorkerAvailability,
|
||||
shutdown_timeout: time::Duration,
|
||||
) {
|
||||
availability.set(false);
|
||||
let mut wrk = MAX_CONNS_COUNTER.with(|conns| Worker {
|
||||
rx,
|
||||
rx2,
|
||||
availability,
|
||||
factories,
|
||||
shutdown_timeout,
|
||||
@ -143,8 +155,12 @@ impl Worker {
|
||||
});
|
||||
|
||||
let mut fut = Vec::new();
|
||||
for factory in &wrk.factories {
|
||||
fut.push(factory.create());
|
||||
for (idx, factory) in wrk.factories.iter().enumerate() {
|
||||
fut.push(factory.create().map(move |res| {
|
||||
res.into_iter()
|
||||
.map(|(t, s)| (idx, t, s))
|
||||
.collect::<Vec<_>>()
|
||||
}));
|
||||
}
|
||||
spawn(
|
||||
future::join_all(fut)
|
||||
@ -152,7 +168,14 @@ impl Worker {
|
||||
error!("Can not start worker: {:?}", e);
|
||||
Arbiter::current().do_send(StopArbiter(0));
|
||||
}).and_then(move |services| {
|
||||
wrk.services.extend(services);
|
||||
for item in services {
|
||||
for (idx, token, service) in item {
|
||||
while token.0 >= wrk.services.len() {
|
||||
wrk.services.push(None);
|
||||
}
|
||||
wrk.services[token.0] = Some((idx, service));
|
||||
}
|
||||
}
|
||||
wrk
|
||||
}),
|
||||
);
|
||||
@ -161,33 +184,42 @@ impl Worker {
|
||||
fn shutdown(&mut self, force: bool) {
|
||||
if force {
|
||||
self.services.iter_mut().for_each(|h| {
|
||||
let _ = h.call((None, ServerMessage::ForceShutdown));
|
||||
if let Some(h) = h {
|
||||
let _ = h.1.call((None, ServerMessage::ForceShutdown));
|
||||
}
|
||||
});
|
||||
} else {
|
||||
let timeout = self.shutdown_timeout;
|
||||
self.services.iter_mut().for_each(move |h| {
|
||||
let _ = h.call((None, ServerMessage::Shutdown(timeout.clone())));
|
||||
if let Some(h) = h {
|
||||
let _ = h.1.call((None, ServerMessage::Shutdown(timeout.clone())));
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
fn check_readiness(&mut self, trace: bool) -> Result<bool, usize> {
|
||||
fn check_readiness(&mut self, trace: bool) -> Result<bool, (Token, usize)> {
|
||||
let mut ready = self.conns.available();
|
||||
let mut failed = None;
|
||||
for (idx, service) in self.services.iter_mut().enumerate() {
|
||||
match service.poll_ready() {
|
||||
Ok(Async::Ready(_)) => {
|
||||
if trace {
|
||||
trace!("Service {:?} is available", self.factories[idx].name());
|
||||
for (token, service) in &mut self.services.iter_mut().enumerate() {
|
||||
if let Some(service) = service {
|
||||
match service.1.poll_ready() {
|
||||
Ok(Async::Ready(_)) => {
|
||||
if trace {
|
||||
trace!(
|
||||
"Service {:?} is available",
|
||||
self.factories[service.0].name(Token(token))
|
||||
);
|
||||
}
|
||||
}
|
||||
Ok(Async::NotReady) => ready = false,
|
||||
Err(_) => {
|
||||
error!(
|
||||
"Service {:?} readiness check returned error, restarting",
|
||||
self.factories[service.0].name(Token(token))
|
||||
);
|
||||
failed = Some((Token(token), service.0));
|
||||
}
|
||||
}
|
||||
Ok(Async::NotReady) => ready = false,
|
||||
Err(_) => {
|
||||
error!(
|
||||
"Service {:?} readiness check returned error, restarting",
|
||||
self.factories[idx].name()
|
||||
);
|
||||
failed = Some(idx);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -203,7 +235,11 @@ enum WorkerState {
|
||||
None,
|
||||
Available,
|
||||
Unavailable(Vec<Conn>),
|
||||
Restarting(usize, Box<Future<Item = BoxedServerService, Error = ()>>),
|
||||
Restarting(
|
||||
usize,
|
||||
Token,
|
||||
Box<Future<Item = Vec<(Token, BoxedServerService)>, Error = ()>>,
|
||||
),
|
||||
Shutdown(Delay, Delay, oneshot::Sender<bool>),
|
||||
}
|
||||
|
||||
@ -212,6 +248,39 @@ impl Future for Worker {
|
||||
type Error = ();
|
||||
|
||||
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
|
||||
// `StopWorker` message handler
|
||||
match self.rx2.poll() {
|
||||
Ok(Async::Ready(Some(StopCommand { graceful, result }))) => {
|
||||
self.availability.set(false);
|
||||
let num = num_connections();
|
||||
if num == 0 {
|
||||
info!("Shutting down worker, 0 connections");
|
||||
let _ = result.send(true);
|
||||
return Ok(Async::Ready(()));
|
||||
} else if graceful {
|
||||
self.shutdown(false);
|
||||
let num = num_connections();
|
||||
if num != 0 {
|
||||
info!("Graceful worker shutdown, {} connections", num);
|
||||
self.state = WorkerState::Shutdown(
|
||||
sleep(time::Duration::from_secs(1)),
|
||||
sleep(self.shutdown_timeout),
|
||||
result,
|
||||
);
|
||||
} else {
|
||||
let _ = result.send(true);
|
||||
return Ok(Async::Ready(()));
|
||||
}
|
||||
} else {
|
||||
info!("Force shutdown worker, {} connections", num);
|
||||
self.shutdown(true);
|
||||
let _ = result.send(false);
|
||||
return Ok(Async::Ready(()));
|
||||
}
|
||||
}
|
||||
_ => (),
|
||||
}
|
||||
|
||||
let state = mem::replace(&mut self.state, WorkerState::None);
|
||||
|
||||
match state {
|
||||
@ -225,7 +294,10 @@ impl Future for Worker {
|
||||
match self.check_readiness(false) {
|
||||
Ok(true) => {
|
||||
let guard = self.conns.get();
|
||||
let _ = self.services[msg.handler.0]
|
||||
let _ = self.services[msg.token.0]
|
||||
.as_mut()
|
||||
.expect("actix net bug")
|
||||
.1
|
||||
.call((Some(guard), ServerMessage::Connect(msg.io)));
|
||||
}
|
||||
Ok(false) => {
|
||||
@ -233,13 +305,14 @@ impl Future for Worker {
|
||||
self.state = WorkerState::Unavailable(conns);
|
||||
return self.poll();
|
||||
}
|
||||
Err(idx) => {
|
||||
Err((token, idx)) => {
|
||||
trace!(
|
||||
"Service {:?} failed, restarting",
|
||||
self.factories[idx].name()
|
||||
self.factories[idx].name(token)
|
||||
);
|
||||
self.state = WorkerState::Restarting(
|
||||
idx,
|
||||
token,
|
||||
self.factories[idx].create(),
|
||||
);
|
||||
return self.poll();
|
||||
@ -253,32 +326,38 @@ impl Future for Worker {
|
||||
self.state = WorkerState::Unavailable(conns);
|
||||
return Ok(Async::NotReady);
|
||||
}
|
||||
Err(idx) => {
|
||||
Err((token, idx)) => {
|
||||
trace!(
|
||||
"Service {:?} failed, restarting",
|
||||
self.factories[idx].name()
|
||||
self.factories[idx].name(token)
|
||||
);
|
||||
self.state = WorkerState::Restarting(idx, self.factories[idx].create());
|
||||
self.state =
|
||||
WorkerState::Restarting(idx, token, self.factories[idx].create());
|
||||
return self.poll();
|
||||
}
|
||||
}
|
||||
}
|
||||
WorkerState::Restarting(idx, mut fut) => {
|
||||
WorkerState::Restarting(idx, token, mut fut) => {
|
||||
match fut.poll() {
|
||||
Ok(Async::Ready(service)) => {
|
||||
trace!(
|
||||
"Service {:?} has been restarted",
|
||||
self.factories[idx].name()
|
||||
);
|
||||
self.services[idx] = service;
|
||||
self.state = WorkerState::Unavailable(Vec::new());
|
||||
Ok(Async::Ready(item)) => {
|
||||
for (token, service) in item {
|
||||
trace!(
|
||||
"Service {:?} has been restarted",
|
||||
self.factories[idx].name(token)
|
||||
);
|
||||
self.services[token.0] = Some((idx, service));
|
||||
self.state = WorkerState::Unavailable(Vec::new());
|
||||
}
|
||||
}
|
||||
Ok(Async::NotReady) => {
|
||||
self.state = WorkerState::Restarting(idx, fut);
|
||||
self.state = WorkerState::Restarting(idx, token, fut);
|
||||
return Ok(Async::NotReady);
|
||||
}
|
||||
Err(_) => {
|
||||
panic!("Can not restart {:?} service", self.factories[idx].name());
|
||||
panic!(
|
||||
"Can not restart {:?} service",
|
||||
self.factories[idx].name(token)
|
||||
);
|
||||
}
|
||||
}
|
||||
return self.poll();
|
||||
@ -317,11 +396,14 @@ impl Future for Worker {
|
||||
loop {
|
||||
match self.rx.poll() {
|
||||
// handle incoming tcp stream
|
||||
Ok(Async::Ready(Some(WorkerCommand::Message(msg)))) => {
|
||||
Ok(Async::Ready(Some(WorkerCommand(msg)))) => {
|
||||
match self.check_readiness(false) {
|
||||
Ok(true) => {
|
||||
let guard = self.conns.get();
|
||||
let _ = self.services[msg.handler.0]
|
||||
let _ = self.services[msg.token.0]
|
||||
.as_mut()
|
||||
.expect("actix net bug")
|
||||
.1
|
||||
.call((Some(guard), ServerMessage::Connect(msg.io)));
|
||||
continue;
|
||||
}
|
||||
@ -330,49 +412,21 @@ impl Future for Worker {
|
||||
self.availability.set(false);
|
||||
self.state = WorkerState::Unavailable(vec![msg]);
|
||||
}
|
||||
Err(idx) => {
|
||||
Err((token, idx)) => {
|
||||
trace!(
|
||||
"Service {:?} failed, restarting",
|
||||
self.factories[idx].name()
|
||||
self.factories[idx].name(token)
|
||||
);
|
||||
self.availability.set(false);
|
||||
self.state = WorkerState::Restarting(
|
||||
idx,
|
||||
token,
|
||||
self.factories[idx].create(),
|
||||
);
|
||||
}
|
||||
}
|
||||
return self.poll();
|
||||
}
|
||||
// `StopWorker` message handler
|
||||
Ok(Async::Ready(Some(WorkerCommand::Stop(graceful, tx)))) => {
|
||||
self.availability.set(false);
|
||||
let num = num_connections();
|
||||
if num == 0 {
|
||||
info!("Shutting down worker, 0 connections");
|
||||
let _ = tx.send(true);
|
||||
return Ok(Async::Ready(()));
|
||||
} else if graceful {
|
||||
self.shutdown(false);
|
||||
let num = num_connections();
|
||||
if num != 0 {
|
||||
info!("Graceful worker shutdown, {} connections", num);
|
||||
break Some(WorkerState::Shutdown(
|
||||
sleep(time::Duration::from_secs(1)),
|
||||
sleep(self.shutdown_timeout),
|
||||
tx,
|
||||
));
|
||||
} else {
|
||||
let _ = tx.send(true);
|
||||
return Ok(Async::Ready(()));
|
||||
}
|
||||
} else {
|
||||
info!("Force shutdown worker, {} connections", num);
|
||||
self.shutdown(true);
|
||||
let _ = tx.send(false);
|
||||
return Ok(Async::Ready(()));
|
||||
}
|
||||
}
|
||||
Ok(Async::NotReady) => {
|
||||
self.state = WorkerState::Available;
|
||||
return Ok(Async::NotReady);
|
||||
@ -383,7 +437,5 @@ impl Future for Worker {
|
||||
}
|
||||
WorkerState::None => panic!(),
|
||||
};
|
||||
|
||||
Ok(Async::NotReady)
|
||||
}
|
||||
}
|
||||
|
@ -27,7 +27,9 @@ pub trait ServiceExt: Service {
|
||||
/// Apply function to specified service and use it as a next service in
|
||||
/// chain.
|
||||
fn apply<S, I, F, R>(
|
||||
self, service: I, f: F,
|
||||
self,
|
||||
service: I,
|
||||
f: F,
|
||||
) -> AndThen<Self, Apply<S, F, R, Self::Response>>
|
||||
where
|
||||
Self: Sized,
|
||||
@ -120,7 +122,9 @@ pub trait ServiceExt: Service {
|
||||
|
||||
pub trait NewServiceExt: NewService {
|
||||
fn apply<S, I, F, R>(
|
||||
self, service: I, f: F,
|
||||
self,
|
||||
service: I,
|
||||
f: F,
|
||||
) -> AndThenNewService<Self, ApplyNewService<S, F, R, Self::Response>>
|
||||
where
|
||||
Self: Sized,
|
||||
|
@ -6,8 +6,8 @@ use tokio_io::{AsyncRead, AsyncWrite};
|
||||
use tokio_openssl::{AcceptAsync, ConnectAsync, SslAcceptorExt, SslConnectorExt, SslStream};
|
||||
|
||||
use super::MAX_CONN_COUNTER;
|
||||
use connector::ConnectionInfo;
|
||||
use counter::{Counter, CounterGuard};
|
||||
use resolver::RequestHost;
|
||||
use service::{NewService, Service};
|
||||
|
||||
/// Support `SSL` connections via openssl package
|
||||
@ -102,113 +102,95 @@ impl<T: AsyncRead + AsyncWrite> Future for OpensslAcceptorServiceFut<T> {
|
||||
}
|
||||
|
||||
/// Openssl connector factory
|
||||
pub struct OpensslConnector<T, Io, E> {
|
||||
pub struct OpensslConnector<R, T, E> {
|
||||
connector: SslConnector,
|
||||
t: PhantomData<T>,
|
||||
io: PhantomData<Io>,
|
||||
_e: PhantomData<E>,
|
||||
_t: PhantomData<(R, T, E)>,
|
||||
}
|
||||
|
||||
impl<T, Io, E> OpensslConnector<T, Io, E> {
|
||||
impl<R, T, E> OpensslConnector<R, T, E> {
|
||||
pub fn new(connector: SslConnector) -> Self {
|
||||
OpensslConnector {
|
||||
connector,
|
||||
t: PhantomData,
|
||||
io: PhantomData,
|
||||
_e: PhantomData,
|
||||
_t: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, Io: AsyncRead + AsyncWrite> OpensslConnector<T, Io, ()> {
|
||||
impl<R: RequestHost, T: AsyncRead + AsyncWrite> OpensslConnector<R, T, ()> {
|
||||
pub fn service(
|
||||
connector: SslConnector,
|
||||
) -> impl Service<
|
||||
Request = (T, ConnectionInfo, Io),
|
||||
Response = (T, ConnectionInfo, SslStream<Io>),
|
||||
Error = Error,
|
||||
> {
|
||||
) -> impl Service<Request = (R, T), Response = (R, SslStream<T>), Error = Error> {
|
||||
OpensslConnectorService {
|
||||
connector: connector,
|
||||
t: PhantomData,
|
||||
io: PhantomData,
|
||||
_t: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, Io, E> Clone for OpensslConnector<T, Io, E> {
|
||||
impl<R, T, E> Clone for OpensslConnector<R, T, E> {
|
||||
fn clone(&self) -> Self {
|
||||
Self {
|
||||
connector: self.connector.clone(),
|
||||
t: PhantomData,
|
||||
io: PhantomData,
|
||||
_e: PhantomData,
|
||||
_t: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, Io: AsyncRead + AsyncWrite, E> NewService for OpensslConnector<T, Io, E> {
|
||||
type Request = (T, ConnectionInfo, Io);
|
||||
type Response = (T, ConnectionInfo, SslStream<Io>);
|
||||
impl<R: RequestHost, T: AsyncRead + AsyncWrite, E> NewService for OpensslConnector<R, T, E> {
|
||||
type Request = (R, T);
|
||||
type Response = (R, SslStream<T>);
|
||||
type Error = Error;
|
||||
type Service = OpensslConnectorService<T, Io>;
|
||||
type Service = OpensslConnectorService<R, T>;
|
||||
type InitError = E;
|
||||
type Future = FutureResult<Self::Service, Self::InitError>;
|
||||
|
||||
fn new_service(&self) -> Self::Future {
|
||||
ok(OpensslConnectorService {
|
||||
connector: self.connector.clone(),
|
||||
t: PhantomData,
|
||||
io: PhantomData,
|
||||
_t: PhantomData,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
pub struct OpensslConnectorService<T, Io> {
|
||||
pub struct OpensslConnectorService<R, T> {
|
||||
connector: SslConnector,
|
||||
t: PhantomData<T>,
|
||||
io: PhantomData<Io>,
|
||||
_t: PhantomData<(R, T)>,
|
||||
}
|
||||
|
||||
impl<T, Io: AsyncRead + AsyncWrite> Service for OpensslConnectorService<T, Io> {
|
||||
type Request = (T, ConnectionInfo, Io);
|
||||
type Response = (T, ConnectionInfo, SslStream<Io>);
|
||||
impl<R: RequestHost, T: AsyncRead + AsyncWrite> Service for OpensslConnectorService<R, T> {
|
||||
type Request = (R, T);
|
||||
type Response = (R, SslStream<T>);
|
||||
type Error = Error;
|
||||
type Future = ConnectAsyncExt<T, Io>;
|
||||
type Future = ConnectAsyncExt<R, T>;
|
||||
|
||||
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
|
||||
Ok(Async::Ready(()))
|
||||
}
|
||||
|
||||
fn call(&mut self, (req, info, stream): Self::Request) -> Self::Future {
|
||||
fn call(&mut self, (req, stream): Self::Request) -> Self::Future {
|
||||
ConnectAsyncExt {
|
||||
fut: SslConnectorExt::connect_async(&self.connector, &info.host, stream),
|
||||
fut: SslConnectorExt::connect_async(&self.connector, req.host(), stream),
|
||||
req: Some(req),
|
||||
host: Some(info),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct ConnectAsyncExt<T, Io> {
|
||||
fut: ConnectAsync<Io>,
|
||||
req: Option<T>,
|
||||
host: Option<ConnectionInfo>,
|
||||
pub struct ConnectAsyncExt<R, T> {
|
||||
req: Option<R>,
|
||||
fut: ConnectAsync<T>,
|
||||
}
|
||||
|
||||
impl<T, Io> Future for ConnectAsyncExt<T, Io>
|
||||
impl<R, T> Future for ConnectAsyncExt<R, T>
|
||||
where
|
||||
Io: AsyncRead + AsyncWrite,
|
||||
R: RequestHost,
|
||||
T: AsyncRead + AsyncWrite,
|
||||
{
|
||||
type Item = (T, ConnectionInfo, SslStream<Io>);
|
||||
type Item = (R, SslStream<T>);
|
||||
type Error = Error;
|
||||
|
||||
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
|
||||
match self.fut.poll()? {
|
||||
Async::Ready(stream) => Ok(Async::Ready((
|
||||
self.req.take().unwrap(),
|
||||
self.host.take().unwrap(),
|
||||
stream,
|
||||
))),
|
||||
Async::Ready(stream) => Ok(Async::Ready((self.req.take().unwrap(), stream))),
|
||||
Async::NotReady => Ok(Async::NotReady),
|
||||
}
|
||||
}
|
||||
|
@ -2,7 +2,7 @@ use std::marker::PhantomData;
|
||||
|
||||
use futures::unsync::mpsc;
|
||||
use futures::{future, Async, Future, Poll, Stream};
|
||||
use tokio::executor::current_thread::spawn;
|
||||
use tokio_current_thread::spawn;
|
||||
|
||||
use super::service::{IntoService, NewService, Service};
|
||||
|
||||
@ -103,6 +103,12 @@ impl<T> TakeItem<T> {
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> Clone for TakeItem<T> {
|
||||
fn clone(&self) -> TakeItem<T> {
|
||||
TakeItem { _t: PhantomData }
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: Stream> NewService for TakeItem<T> {
|
||||
type Request = T;
|
||||
type Response = (Option<T::Item>, T);
|
||||
@ -121,6 +127,12 @@ pub struct TakeItemService<T> {
|
||||
_t: PhantomData<T>,
|
||||
}
|
||||
|
||||
impl<T> Clone for TakeItemService<T> {
|
||||
fn clone(&self) -> TakeItemService<T> {
|
||||
TakeItemService { _t: PhantomData }
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: Stream> Service for TakeItemService<T> {
|
||||
type Request = T;
|
||||
type Response = (Option<T::Item>, T);
|
||||
|
@ -10,7 +10,7 @@ use super::service::{NewService, Service};
|
||||
use super::Never;
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct LowResTimer(Cell<Inner>);
|
||||
pub struct LowResTime(Cell<Inner>);
|
||||
|
||||
#[derive(Debug)]
|
||||
struct Inner {
|
||||
@ -27,28 +27,28 @@ impl Inner {
|
||||
}
|
||||
}
|
||||
|
||||
impl LowResTimer {
|
||||
pub fn with(resolution: Duration) -> LowResTimer {
|
||||
LowResTimer(Cell::new(Inner::new(resolution)))
|
||||
impl LowResTime {
|
||||
pub fn with(resolution: Duration) -> LowResTime {
|
||||
LowResTime(Cell::new(Inner::new(resolution)))
|
||||
}
|
||||
|
||||
pub fn timer(&self) -> LowResTimerService {
|
||||
LowResTimerService(self.0.clone())
|
||||
pub fn timer(&self) -> LowResTimeService {
|
||||
LowResTimeService(self.0.clone())
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for LowResTimer {
|
||||
impl Default for LowResTime {
|
||||
fn default() -> Self {
|
||||
LowResTimer(Cell::new(Inner::new(Duration::from_secs(1))))
|
||||
LowResTime(Cell::new(Inner::new(Duration::from_secs(1))))
|
||||
}
|
||||
}
|
||||
|
||||
impl NewService for LowResTimer {
|
||||
impl NewService for LowResTime {
|
||||
type Request = ();
|
||||
type Response = Instant;
|
||||
type Error = Never;
|
||||
type InitError = Never;
|
||||
type Service = LowResTimerService;
|
||||
type Service = LowResTimeService;
|
||||
type Future = FutureResult<Self::Service, Self::InitError>;
|
||||
|
||||
fn new_service(&self) -> Self::Future {
|
||||
@ -57,16 +57,16 @@ impl NewService for LowResTimer {
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct LowResTimerService(Cell<Inner>);
|
||||
pub struct LowResTimeService(Cell<Inner>);
|
||||
|
||||
impl LowResTimerService {
|
||||
pub fn with(resolution: Duration) -> LowResTimerService {
|
||||
LowResTimerService(Cell::new(Inner::new(resolution)))
|
||||
impl LowResTimeService {
|
||||
pub fn with(resolution: Duration) -> LowResTimeService {
|
||||
LowResTimeService(Cell::new(Inner::new(resolution)))
|
||||
}
|
||||
|
||||
/// Get current time. This function has to be called from
|
||||
/// future's poll method, otherwise it panics.
|
||||
pub fn now(&mut self) -> Instant {
|
||||
pub fn now(&self) -> Instant {
|
||||
let cur = self.0.borrow().current.clone();
|
||||
if let Some(cur) = cur {
|
||||
cur
|
||||
@ -88,7 +88,7 @@ impl LowResTimerService {
|
||||
}
|
||||
}
|
||||
|
||||
impl Service for LowResTimerService {
|
||||
impl Service for LowResTimeService {
|
||||
type Request = ();
|
||||
type Response = Instant;
|
||||
type Error = Never;
|
162
src/timeout.rs
Normal file
162
src/timeout.rs
Normal file
@ -0,0 +1,162 @@
|
||||
//! Service that applies a timeout to requests.
|
||||
//!
|
||||
//! If the response does not complete within the specified timeout, the response
|
||||
//! will be aborted.
|
||||
use std::fmt;
|
||||
use std::time::Duration;
|
||||
|
||||
use futures::{Async, Future, Poll};
|
||||
use tokio_timer::{clock, Delay};
|
||||
|
||||
use service::{NewService, Service};
|
||||
|
||||
/// Applies a timeout to requests.
|
||||
#[derive(Debug)]
|
||||
pub struct Timeout<T: NewService + Clone> {
|
||||
inner: T,
|
||||
timeout: Duration,
|
||||
}
|
||||
|
||||
/// Timeout error
|
||||
pub enum TimeoutError<E> {
|
||||
/// Service error
|
||||
Service(E),
|
||||
/// Service call timeout
|
||||
Timeout,
|
||||
}
|
||||
|
||||
impl<E: fmt::Debug> fmt::Debug for TimeoutError<E> {
|
||||
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
|
||||
match self {
|
||||
TimeoutError::Service(e) => write!(f, "TimeoutError::Service({:?})", e),
|
||||
TimeoutError::Timeout => write!(f, "TimeoutError::Timeout"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> Timeout<T>
|
||||
where
|
||||
T: NewService + Clone,
|
||||
{
|
||||
pub fn new(timeout: Duration, inner: T) -> Self {
|
||||
Timeout { inner, timeout }
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> NewService for Timeout<T>
|
||||
where
|
||||
T: NewService + Clone,
|
||||
{
|
||||
type Request = T::Request;
|
||||
type Response = T::Response;
|
||||
type Error = TimeoutError<T::Error>;
|
||||
type InitError = T::InitError;
|
||||
type Service = TimeoutService<T::Service>;
|
||||
type Future = TimeoutFut<T>;
|
||||
|
||||
fn new_service(&self) -> Self::Future {
|
||||
TimeoutFut {
|
||||
fut: self.inner.new_service(),
|
||||
timeout: self.timeout.clone(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// `Timeout` response future
|
||||
#[derive(Debug)]
|
||||
pub struct TimeoutFut<T: NewService> {
|
||||
fut: T::Future,
|
||||
timeout: Duration,
|
||||
}
|
||||
|
||||
impl<T> Future for TimeoutFut<T>
|
||||
where
|
||||
T: NewService,
|
||||
{
|
||||
type Item = TimeoutService<T::Service>;
|
||||
type Error = T::InitError;
|
||||
|
||||
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
|
||||
let service = try_ready!(self.fut.poll());
|
||||
Ok(Async::Ready(TimeoutService::new(self.timeout, service)))
|
||||
}
|
||||
}
|
||||
|
||||
/// Applies a timeout to requests.
|
||||
#[derive(Debug)]
|
||||
pub struct TimeoutService<T> {
|
||||
inner: T,
|
||||
timeout: Duration,
|
||||
}
|
||||
|
||||
impl<T> TimeoutService<T> {
|
||||
pub fn new(timeout: Duration, inner: T) -> Self {
|
||||
TimeoutService { inner, timeout }
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> Clone for TimeoutService<T>
|
||||
where
|
||||
T: Clone,
|
||||
{
|
||||
fn clone(&self) -> Self {
|
||||
TimeoutService {
|
||||
inner: self.inner.clone(),
|
||||
timeout: self.timeout,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> Service for TimeoutService<T>
|
||||
where
|
||||
T: Service,
|
||||
{
|
||||
type Request = T::Request;
|
||||
type Response = T::Response;
|
||||
type Error = TimeoutError<T::Error>;
|
||||
type Future = TimeoutServiceResponse<T>;
|
||||
|
||||
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
|
||||
self.inner
|
||||
.poll_ready()
|
||||
.map_err(|e| TimeoutError::Service(e))
|
||||
}
|
||||
|
||||
fn call(&mut self, request: Self::Request) -> Self::Future {
|
||||
TimeoutServiceResponse {
|
||||
fut: self.inner.call(request),
|
||||
sleep: Delay::new(clock::now() + self.timeout),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// `TimeoutService` response future
|
||||
#[derive(Debug)]
|
||||
pub struct TimeoutServiceResponse<T: Service> {
|
||||
fut: T::Future,
|
||||
sleep: Delay,
|
||||
}
|
||||
|
||||
impl<T> Future for TimeoutServiceResponse<T>
|
||||
where
|
||||
T: Service,
|
||||
{
|
||||
type Item = T::Response;
|
||||
type Error = TimeoutError<T::Error>;
|
||||
|
||||
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
|
||||
// First, try polling the future
|
||||
match self.fut.poll() {
|
||||
Ok(Async::Ready(v)) => return Ok(Async::Ready(v)),
|
||||
Ok(Async::NotReady) => {}
|
||||
Err(e) => return Err(TimeoutError::Service(e)),
|
||||
}
|
||||
|
||||
// Now check the sleep
|
||||
match self.sleep.poll() {
|
||||
Ok(Async::NotReady) => Ok(Async::NotReady),
|
||||
Ok(Async::Ready(_)) => Err(TimeoutError::Timeout),
|
||||
Err(_) => Err(TimeoutError::Timeout),
|
||||
}
|
||||
}
|
||||
}
|
Reference in New Issue
Block a user