prep release

added docs for trait Service, trait Transform

.travis.yml

@@ -37,36 +37,13 @@ script:
   - |
     if [[ "$TRAVIS_RUST_VERSION" != "nightly-2019-03-02" ]]; then
     cargo clean
-    cargo test --features="ssl,tls,rust-tls" -- --nocapture
-    cd actix-codec && cargo test && cd ..
-    cd actix-service && cargo test && cd ..
-    cd actix-server && cargo test --all-features -- --nocapture && cd ..
-    cd actix-rt && cargo test && cd ..
-    cd actix-connect && cargo test && cd ..
-    cd actix-utils && cargo test && cd ..
-    cd router && cargo test && cd ..
+    cargo test --all --all-features -- --nocapture
     fi
 
 after_success:
   - |
     if [[ "$TRAVIS_RUST_VERSION" == "nightly-2019-03-02" ]]; then
-    #cd actix-service && cargo tarpaulin --out Xml && bash <(curl -s https://codecov.io/bash) && cd ..
-    #cd actix-rt && cargo tarpaulin --out Xml && bash <(curl -s https://codecov.io/bash) && cd ..
-    #cd actix-connect && cargo tarpaulin --out Xml && bash <(curl -s https://codecov.io/bash) && cd ..
-    #cd actix-codec && cargo tarpaulin --out Xml && bash <(curl -s https://codecov.io/bash) && cd ..
-    #cd actix-server && cargo tarpaulin --out Xml && bash <(curl -s https://codecov.io/bash) && cd ..
-    #cd actix-utils && cargo tarpaulin --out Xml && bash <(curl -s https://codecov.io/bash) && cd ..
-    #cd router && cargo tarpaulin --out Xml && bash <(curl -s https://codecov.io/bash) && cd ..
-
-    cd actix-service && cargo tarpaulin --out Xml && cd ..
-    #cd actix-rt && cargo tarpaulin --out Xml && cd ..
-    cd actix-connect && cargo tarpaulin --out Xml && cd ..
-    #cd actix-codec && cargo tarpaulin --out Xml && cd ..
-    #cd actix-server && cargo tarpaulin --out Xml && cd ..
-    cd actix-utils && cargo tarpaulin --out Xml && cd ..
-    cd router && cargo tarpaulin --out Xml && cd ..
-
-    # cargo tarpaulin --all --all-features --out Xml
+    taskset -c 0 cargo tarpaulin --all --all-features --out Xml
     echo "Uploaded code coverage"
     bash <(curl -s https://codecov.io/bash)
     fi

Cargo.toml

@@ -22,6 +22,7 @@ members = [
   "actix-server",
   "actix-server-config",
   "actix-test-server",
+  "actix-threadpool",
   "actix-utils",
   "router",
 ]

actix-codec/CHANGES.md

@@ -1,6 +1,11 @@
 # Changes
 
-## [0.1.0] - 2019-03-06
+## [0.1.2] - 2019-03-27
+
+* Added `Framed::map_io()` method.
+
+
+## [0.1.1] - 2019-03-06
 
 * Added `FramedParts::with_read_buffer()` method.
 

actix-codec/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-codec"
-version = "0.1.1"
+version = "0.1.2"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Utilities for encoding and decoding frames"
 keywords = ["network", "framework", "async", "futures"]
@@ -11,15 +11,15 @@ categories = ["network-programming", "asynchronous"]
 license = "MIT/Apache-2.0"
 exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
 edition = "2018"
-workspace = "../"
+workspace = ".."
 
 [lib]
 name = "actix_codec"
 path = "src/lib.rs"
 
 [dependencies]
-bytes = "0.4"
+bytes = "0.4.12"
 futures = "0.1.24"
-tokio-io = "0.1"
-tokio-codec = "0.1"
+tokio-io = "0.1.12"
+tokio-codec = "0.1.1"
 log = "0.4"

actix-codec/src/framed.rs

@@ -167,6 +167,22 @@ impl<T, U> Framed<T, U> {
         }
     }
 
+    /// Consume the `Frame`, returning `Frame` with different io.
+    pub fn map_io<F, T2>(self, f: F) -> Framed<T2, U>
+    where
+        F: Fn(T) -> T2,
+    {
+        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(f(inner.0), inner.1), 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

actix-connect/CHANGES.md

@@ -1,5 +1,36 @@
 # Changes
 
+## [0.1.4] - 2019-04-12
+
+### Changed
+
+* Do not start default resolver immediately for default connector.
+
+
+## [0.1.3] - 2019-04-11
+
+### Changed
+
+* Start trust-dns default resolver on first use
+
+## [0.1.2] - 2019-04-04
+
+### Added
+
+* Log error if dns system config could not be loaded.
+
+### Changed
+
+* Rename connect Connector to TcpConnector #10
+
+
+## [0.1.1] - 2019-03-15
+
+### Fixed
+
+* Fix error handling for single address
+
+
 ## [0.1.0] - 2019-03-14
 
 * Refactor resolver and connector services

actix-connect/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-connect"
-version = "0.1.0"
+version = "0.1.4"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix Connector - tcp connector service"
 keywords = ["network", "framework", "async", "futures"]
@@ -30,9 +30,9 @@ ssl = ["openssl", "tokio-openssl"]
 uri = ["http"]
 
 [dependencies]
-actix-service = "0.3.4"
-actix-codec = "0.1.1"
-actix-utils = "0.3.4"
+actix-service = "0.3.6"
+actix-codec = "0.1.2"
+actix-utils = "0.3.5"
 derive_more = "0.14.0"
 either = "1.5.1"
 futures = "0.1.25"
@@ -40,7 +40,7 @@ http = { version = "0.1.16", optional = true }
 log = "0.4"
 tokio-tcp = "0.1.3"
 tokio-current-thread = "0.1.5"
-trust-dns-resolver = { version="0.11.0-alpha.2", default-features = false }
+trust-dns-resolver = { version="0.11.0-alpha.3", default-features = false }
 
 # openssl
 openssl = { version="0.10", optional = true }
@@ -48,8 +48,7 @@ tokio-openssl = { version="0.3", optional = true }
 
 [dev-dependencies]
 bytes = "0.4"
-actix-connect = { path=".", features=["ssl"] }
-actix-test-server = { version="0.2.0", features=["ssl"] }
+actix-test-server = { version="0.2.1", features=["ssl"] }
 actix-server-config = "0.1.0"
 actix-utils = "0.3.4"
 tokio-tcp = "0.1"

actix-connect/src/connector.rs

@@ -12,54 +12,54 @@ use super::error::ConnectError;
 
 /// Tcp connector service factory
 #[derive(Debug)]
-pub struct ConnectorFactory<T>(PhantomData<T>);
+pub struct TcpConnectorFactory<T>(PhantomData<T>);
 
-impl<T> ConnectorFactory<T> {
+impl<T> TcpConnectorFactory<T> {
     pub fn new() -> Self {
-        ConnectorFactory(PhantomData)
+        TcpConnectorFactory(PhantomData)
     }
 }
 
-impl<T> Clone for ConnectorFactory<T> {
+impl<T> Clone for TcpConnectorFactory<T> {
     fn clone(&self) -> Self {
-        ConnectorFactory(PhantomData)
+        TcpConnectorFactory(PhantomData)
     }
 }
 
-impl<T: Address> NewService for ConnectorFactory<T> {
+impl<T: Address> NewService for TcpConnectorFactory<T> {
     type Request = Connect<T>;
     type Response = Connection<T, TcpStream>;
     type Error = ConnectError;
-    type Service = Connector<T>;
+    type Service = TcpConnector<T>;
     type InitError = ();
     type Future = FutureResult<Self::Service, Self::InitError>;
 
     fn new_service(&self, _: &()) -> Self::Future {
-        ok(Connector(PhantomData))
+        ok(TcpConnector(PhantomData))
     }
 }
 
 /// Tcp connector service
 #[derive(Debug)]
-pub struct Connector<T>(PhantomData<T>);
+pub struct TcpConnector<T>(PhantomData<T>);
 
-impl<T> Connector<T> {
+impl<T> TcpConnector<T> {
     pub fn new() -> Self {
-        Connector(PhantomData)
+        TcpConnector(PhantomData)
     }
 }
 
-impl<T> Clone for Connector<T> {
+impl<T> Clone for TcpConnector<T> {
     fn clone(&self) -> Self {
-        Connector(PhantomData)
+        TcpConnector(PhantomData)
     }
 }
 
-impl<T: Address> Service for Connector<T> {
+impl<T: Address> Service for TcpConnector<T> {
     type Request = Connect<T>;
     type Response = Connection<T, TcpStream>;
     type Error = ConnectError;
-    type Future = Either<ConnectorResponse<T>, FutureResult<Self::Response, Self::Error>>;
+    type Future = Either<TcpConnectorResponse<T>, FutureResult<Self::Response, Self::Error>>;
 
     fn poll_ready(&mut self) -> Poll<(), Self::Error> {
         Ok(Async::Ready(()))
@@ -70,7 +70,7 @@ impl<T: Address> Service for Connector<T> {
         let Connect { req, addr, .. } = req;
 
         if let Some(addr) = addr {
-            Either::A(ConnectorResponse::new(req, port, addr))
+            Either::A(TcpConnectorResponse::new(req, port, addr))
         } else {
             error!("TCP connector: got unresolved address");
             Either::B(err(ConnectError::Unresolverd))
@@ -80,19 +80,19 @@ impl<T: Address> Service for Connector<T> {
 
 #[doc(hidden)]
 /// Tcp stream connector response future
-pub struct ConnectorResponse<T> {
+pub struct TcpConnectorResponse<T> {
     req: Option<T>,
     port: u16,
     addrs: Option<VecDeque<SocketAddr>>,
     stream: Option<ConnectFuture>,
 }
 
-impl<T: Address> ConnectorResponse<T> {
+impl<T: Address> TcpConnectorResponse<T> {
     pub fn new(
         req: T,
         port: u16,
         addr: either::Either<SocketAddr, VecDeque<SocketAddr>>,
-    ) -> ConnectorResponse<T> {
+    ) -> TcpConnectorResponse<T> {
         trace!(
             "TCP connector - connecting to {:?} port:{}",
             req.host(),
@@ -100,13 +100,13 @@ impl<T: Address> ConnectorResponse<T> {
         );
 
         match addr {
-            either::Either::Left(addr) => ConnectorResponse {
+            either::Either::Left(addr) => TcpConnectorResponse {
                 req: Some(req),
                 port,
                 addrs: None,
                 stream: Some(TcpStream::connect(&addr)),
             },
-            either::Either::Right(addrs) => ConnectorResponse {
+            either::Either::Right(addrs) => TcpConnectorResponse {
                 req: Some(req),
                 port,
                 addrs: Some(addrs),
@@ -116,7 +116,7 @@ impl<T: Address> ConnectorResponse<T> {
     }
 }
 
-impl<T: Address> Future for ConnectorResponse<T> {
+impl<T: Address> Future for TcpConnectorResponse<T> {
     type Item = Connection<T, TcpStream>;
     type Error = ConnectError;
 
@@ -140,7 +140,7 @@ impl<T: Address> Future for ConnectorResponse<T> {
                             self.req.as_ref().unwrap().host(),
                             self.port,
                         );
-                        if self.addrs.as_ref().unwrap().is_empty() {
+                        if self.addrs.is_none() || self.addrs.as_ref().unwrap().is_empty() {
                             return Err(err.into());
                         }
                     }

actix-connect/src/lib.rs

@@ -8,6 +8,8 @@
 #[macro_use]
 extern crate log;
 
+use std::cell::RefCell;
+
 mod connect;
 mod connector;
 mod error;
@@ -17,17 +19,28 @@ pub mod ssl;
 #[cfg(feature = "uri")]
 mod uri;
 
+pub use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
+pub use trust_dns_resolver::system_conf::read_system_conf;
 pub use trust_dns_resolver::{error::ResolveError, AsyncResolver};
 
 pub use self::connect::{Address, Connect, Connection};
-pub use self::connector::{Connector, ConnectorFactory};
+pub use self::connector::{TcpConnector, TcpConnectorFactory};
 pub use self::error::ConnectError;
 pub use self::resolver::{Resolver, ResolverFactory};
 
 use actix_service::{NewService, Service, ServiceExt};
 use tokio_tcp::TcpStream;
-use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
-use trust_dns_resolver::system_conf::read_system_conf;
+
+#[doc(hidden)]
+#[deprecated(since = "0.1.2", note = "please use `actix_connect::TcpConnector`")]
+pub type Connector<T> = TcpConnector<T>;
+
+#[doc(hidden)]
+#[deprecated(
+    since = "0.1.2",
+    note = "please use `actix_connect::TcpConnectorFactory`"
+)]
+pub type ConnectorFactory<T> = TcpConnectorFactory<T>;
 
 pub fn start_resolver(cfg: ResolverConfig, opts: ResolverOpts) -> AsyncResolver {
     let (resolver, bg) = AsyncResolver::new(cfg, opts);
@@ -35,16 +48,34 @@ pub fn start_resolver(cfg: ResolverConfig, opts: ResolverOpts) -> AsyncResolver
     resolver
 }
 
-pub fn start_default_resolver() -> AsyncResolver {
-    let (cfg, opts) = if let Ok((cfg, opts)) = read_system_conf() {
-        (cfg, opts)
-    } else {
-        (ResolverConfig::default(), ResolverOpts::default())
-    };
+thread_local! {
+    static DEFAULT_RESOLVER: RefCell<Option<AsyncResolver>> = RefCell::new(None);
+}
 
-    let (resolver, bg) = AsyncResolver::new(cfg, opts);
-    tokio_current_thread::spawn(bg);
-    resolver
+pub(crate) fn get_default_resolver() -> AsyncResolver {
+    DEFAULT_RESOLVER.with(|cell| {
+        if let Some(ref resolver) = *cell.borrow() {
+            return resolver.clone();
+        }
+
+        let (cfg, opts) = match read_system_conf() {
+            Ok((cfg, opts)) => (cfg, opts),
+            Err(e) => {
+                log::error!("TRust-DNS can not load system config: {}", e);
+                (ResolverConfig::default(), ResolverOpts::default())
+            }
+        };
+
+        let (resolver, bg) = AsyncResolver::new(cfg, opts);
+        tokio_current_thread::spawn(bg);
+
+        *cell.borrow_mut() = Some(resolver.clone());
+        resolver
+    })
+}
+
+pub fn start_default_resolver() -> AsyncResolver {
+    get_default_resolver()
 }
 
 /// Create tcp connector service
@@ -52,7 +83,7 @@ pub fn new_connector<T: Address>(
     resolver: AsyncResolver,
 ) -> impl Service<Request = Connect<T>, Response = Connection<T, TcpStream>, Error = ConnectError>
          + Clone {
-    Resolver::new(resolver).and_then(Connector::new())
+    Resolver::new(resolver).and_then(TcpConnector::new())
 }
 
 /// Create tcp connector service
@@ -64,14 +95,14 @@ pub fn new_connector_factory<T: Address>(
     Error = ConnectError,
     InitError = (),
 > + Clone {
-    ResolverFactory::new(resolver).and_then(ConnectorFactory::new())
+    ResolverFactory::new(resolver).and_then(TcpConnectorFactory::new())
 }
 
 /// Create connector service with default parameters
 pub fn default_connector<T: Address>(
 ) -> impl Service<Request = Connect<T>, Response = Connection<T, TcpStream>, Error = ConnectError>
          + Clone {
-    Resolver::new(start_default_resolver()).and_then(Connector::new())
+    Resolver::default().and_then(TcpConnector::new())
 }
 
 /// Create connector service factory with default parameters
@@ -81,5 +112,5 @@ pub fn default_connector_factory<T: Address>() -> impl NewService<
     Error = ConnectError,
     InitError = (),
 > + Clone {
-    ResolverFactory::new(start_default_resolver()).and_then(ConnectorFactory::new())
+    ResolverFactory::default().and_then(TcpConnectorFactory::new())
 }

actix-connect/src/resolver.rs

@@ -10,10 +10,11 @@ use trust_dns_resolver::{AsyncResolver, Background};
 
 use crate::connect::{Address, Connect};
 use crate::error::ConnectError;
+use crate::get_default_resolver;
 
 /// DNS Resolver Service factory
 pub struct ResolverFactory<T> {
-    resolver: AsyncResolver,
+    resolver: Option<AsyncResolver>,
     _t: PhantomData<T>,
 }
 
@@ -21,13 +22,18 @@ impl<T> ResolverFactory<T> {
     /// Create new resolver instance with custom configuration and options.
     pub fn new(resolver: AsyncResolver) -> Self {
         ResolverFactory {
-            resolver,
+            resolver: Some(resolver),
             _t: PhantomData,
         }
     }
+}
 
-    pub fn resolver(&self) -> &AsyncResolver {
-        &self.resolver
+impl<T> Default for ResolverFactory<T> {
+    fn default() -> Self {
+        ResolverFactory {
+            resolver: None,
+            _t: PhantomData,
+        }
     }
 }
 
@@ -58,7 +64,7 @@ impl<T: Address> NewService for ResolverFactory<T> {
 
 /// DNS Resolver Service
 pub struct Resolver<T> {
-    resolver: AsyncResolver,
+    resolver: Option<AsyncResolver>,
     _t: PhantomData<T>,
 }
 
@@ -66,7 +72,16 @@ impl<T> Resolver<T> {
     /// Create new resolver instance with custom configuration and options.
     pub fn new(resolver: AsyncResolver) -> Self {
         Resolver {
-            resolver,
+            resolver: Some(resolver),
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T> Default for Resolver<T> {
+    fn default() -> Self {
+        Resolver {
+            resolver: None,
             _t: PhantomData,
         }
     }
@@ -100,7 +115,10 @@ impl<T: Address> Service for Resolver<T> {
                 Either::B(ok(req))
             } else {
                 trace!("DNS resolver: resolving host {:?}", req.host());
-                Either::A(ResolverFuture::new(req, &self.resolver))
+                if self.resolver.is_none() {
+                    self.resolver = Some(get_default_resolver());
+                }
+                Either::A(ResolverFuture::new(req, self.resolver.as_ref().unwrap()))
             }
         }
     }
@@ -153,7 +171,9 @@ impl<T: Address> Future for ResolverFuture<T> {
                     req.host(),
                     addrs
                 );
-                if addrs.len() == 1 {
+                if addrs.is_empty() {
+                    Err(ConnectError::NoRecords)
+                } else if addrs.len() == 1 {
                     req.addr = Some(either::Either::Left(addrs.pop_front().unwrap()));
                     Ok(Async::Ready(req))
                 } else {

actix-connect/src/uri.rs

@@ -26,6 +26,7 @@ fn port(scheme: Option<&str>) -> Option<u16> {
             "wss" => Some(443),
             "amqp" => Some(5672),
             "amqps" => Some(5671),
+            "sb" => Some(5671),
             "mqtt" => Some(1883),
             "mqtts" => Some(8883),
             _ => None,

actix-connect/tests/test_connect.rs

@@ -9,6 +9,7 @@ use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
 
 use actix_connect::{default_connector, Connect};
 
+#[cfg(feature = "ssl")]
 #[test]
 fn test_string() {
     let mut srv = TestServer::with(|| {
@@ -19,11 +20,9 @@ fn test_string() {
         })
     });
 
-    let mut conn = srv
-        .block_on(lazy(|| Ok::<_, ()>(default_connector())))
-        .unwrap();
+    let mut conn = default_connector();
     let addr = format!("localhost:{}", srv.port());
-    let con = srv.block_on(conn.call(addr.into())).unwrap();
+    let con = srv.run_on(move || conn.call(addr.into())).unwrap();
     assert_eq!(con.peer_addr().unwrap(), srv.addr());
 }
 
@@ -43,13 +42,22 @@ fn test_static_str() {
         ))
         .unwrap();
     let mut conn = srv
-        .block_on(lazy(|| Ok::<_, ()>(actix_connect::new_connector(resolver))))
+        .block_on(lazy(|| {
+            Ok::<_, ()>(actix_connect::new_connector(resolver.clone()))
+        }))
         .unwrap();
 
     let con = srv
         .block_on(conn.call(Connect::with("10", srv.addr())))
         .unwrap();
     assert_eq!(con.peer_addr().unwrap(), srv.addr());
+
+    let connect = Connect::new(srv.host().to_owned());
+    let mut conn = srv
+        .block_on(lazy(|| Ok::<_, ()>(actix_connect::new_connector(resolver))))
+        .unwrap();
+    let con = srv.block_on(conn.call(connect));
+    assert!(con.is_err());
 }
 
 #[test]
@@ -83,6 +91,7 @@ fn test_new_service() {
     assert_eq!(con.peer_addr().unwrap(), srv.addr());
 }
 
+#[cfg(feature = "ssl")]
 #[test]
 fn test_uri() {
     let mut srv = TestServer::with(|| {
@@ -93,10 +102,8 @@ fn test_uri() {
         })
     });
 
-    let mut conn = srv
-        .block_on(lazy(|| Ok::<_, ()>(default_connector())))
-        .unwrap();
+    let mut conn = default_connector();
     let addr = Uri::try_from(format!("https://localhost:{}", srv.port())).unwrap();
-    let con = srv.block_on(conn.call(addr.into())).unwrap();
+    let con = srv.run_on(move || conn.call(addr.into())).unwrap();
     assert_eq!(con.peer_addr().unwrap(), srv.addr());
 }

actix-rt/CHANGES.md

@@ -1,5 +1,11 @@
 # Changes
 
+## [0.2.2] - 2019-03-28
+
+### Changed
+
+* Moved `blocking` module to `actix-threadpool` crate
+
 ## [0.2.1] - 2019-03-11
 
 ### Added

actix-rt/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-rt"
-version = "0.2.1"
+version = "0.2.2"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix runtime"
 keywords = ["network", "framework", "async", "futures"]
@@ -11,21 +11,15 @@ categories = ["network-programming", "asynchronous"]
 license = "MIT/Apache-2.0"
 exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
 edition = "2018"
-workspace = "../"
+workspace = ".."
 
 [lib]
 name = "actix_rt"
 path = "src/lib.rs"
 
 [dependencies]
-bytes = "0.4"
-derive_more = "0.14"
+actix-threadpool = "0.1.0"
 futures = "0.1.25"
-parking_lot = "0.7"
-lazy_static = "1.2"
-log = "0.4"
-num_cpus = "1.10"
-threadpool = "1.7"
 tokio-current-thread = "0.1"
 tokio-executor = "0.1.5"
 tokio-reactor = "0.1.7"

actix-rt/src/builder.rs

@@ -86,8 +86,7 @@ impl Builder {
         let (stop_tx, stop) = channel();
         let (sys_sender, sys_receiver) = unbounded();
 
-        let arbiter = Arbiter::new_system();
-        let system = System::construct(sys_sender, arbiter.clone(), self.stop_on_panic);
+        let system = System::construct(sys_sender, Arbiter::new_system(), self.stop_on_panic);
 
         // system arbiter
         let arb = SystemArbiter::new(stop_tx, sys_receiver);

actix-rt/src/lib.rs

@@ -1,7 +1,6 @@
 //! A runtime implementation that runs everything on the current thread.
 
 mod arbiter;
-pub mod blocking;
 mod builder;
 mod runtime;
 mod system;
@@ -11,6 +10,9 @@ pub use self::builder::{Builder, SystemRunner};
 pub use self::runtime::Runtime;
 pub use self::system::System;
 
+#[doc(hidden)]
+pub use actix_threadpool as blocking;
+
 /// Spawns a future on the current arbiter.
 ///
 /// # Panics

actix-server/CHANGES.md

@@ -1,5 +1,12 @@
 # Changes
 
+## [0.4.2] - 2019-03-30
+
+### Fixed
+
+* Fix SIGINT force shutdown
+
+
 ## [0.4.1] - 2019-03-14
 
 ### Added

actix-server/Cargo.toml

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

actix-server/src/builder.rs

@@ -328,7 +328,7 @@ impl ServerBuilder {
                 self.accept.send(Command::Stop);
 
                 // stop workers
-                if !self.workers.is_empty() {
+                if !self.workers.is_empty() && graceful {
                     spawn(
                         futures_unordered(
                             self.workers

actix-service/CHANGES.md

@@ -1,5 +1,19 @@
 # Changes
 
+## [0.3.6] - 2019-04-07
+
+### Changed
+
+* Poll boxed service call result immediately
+
+
+## [0.3.5] - 2019-03-29
+
+### Added
+
+* Add `impl<S: Service> Service for Rc<RefCell<S>>`
+
+
 ## [0.3.4] - 2019-03-12
 
 ### Added

actix-service/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-service"
-version = "0.3.4"
+version = "0.3.6"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix Service"
 keywords = ["network", "framework", "async", "futures"]

actix-service/src/boxed.rs

@@ -1,12 +1,14 @@
+use futures::future::{err, ok, Either, FutureResult};
+use futures::{Async, Future, IntoFuture, Poll};
+
 use crate::{NewService, Service};
-use futures::{Future, IntoFuture, Poll};
 
 pub type BoxedService<Req, Res, Err> = Box<
     Service<
         Request = Req,
         Response = Res,
         Error = Err,
-        Future = Box<Future<Item = Res, Error = Err>>,
+        Future = Either<FutureResult<Res, Err>, Box<Future<Item = Res, Error = Err>>>,
     >,
 >;
 
@@ -125,13 +127,21 @@ where
     type Request = Req;
     type Response = Res;
     type Error = Err;
-    type Future = Box<Future<Item = Self::Response, Error = Self::Error>>;
+    type Future = Either<
+        FutureResult<Self::Response, Self::Error>,
+        Box<Future<Item = Self::Response, Error = Self::Error>>,
+    >;
 
     fn poll_ready(&mut self) -> Poll<(), Self::Error> {
         self.0.poll_ready()
     }
 
     fn call(&mut self, req: Self::Request) -> Self::Future {
-        Box::new(self.0.call(req))
+        let mut fut = self.0.call(req);
+        match fut.poll() {
+            Ok(Async::Ready(res)) => Either::A(ok(res)),
+            Err(e) => Either::A(err(e)),
+            Ok(Async::NotReady) => Either::B(Box::new(fut)),
+        }
     }
 }

actix-service/src/lib.rs

@@ -1,3 +1,4 @@
+use std::cell::RefCell;
 use std::rc::Rc;
 use std::sync::Arc;
 
@@ -391,6 +392,24 @@ where
     }
 }
 
+impl<S> Service for Rc<RefCell<S>>
+where
+    S: Service,
+{
+    type Request = S::Request;
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = S::Future;
+
+    fn poll_ready(&mut self) -> Poll<(), S::Error> {
+        self.borrow_mut().poll_ready()
+    }
+
+    fn call(&mut self, request: Self::Request) -> S::Future {
+        self.borrow_mut().call(request)
+    }
+}
+
 impl<S, C> NewService<C> for Rc<S>
 where
     S: NewService<C>,

actix-service/src/transform.rs

@@ -6,9 +6,16 @@ use futures::{Async, Future, IntoFuture, Poll};
 use crate::transform_err::{TransformFromErr, TransformMapInitErr};
 use crate::{IntoNewService, NewService, Service};
 
-/// `Transform` service factory.
+/// The `Transform` trait defines the interface of a Service factory.  `Transform`
+/// is often implemented for middleware, defining how to manufacture a
+/// middleware Service.  A Service that is manufactured by the factory takes
+/// the Service that follows it during execution as a parameter, assuming
+/// ownership of the next Service.  A Service can be a variety of types, such
+/// as (but not limited to) another middleware Service, an extractor Service,
+/// other helper Services, or the request handler endpoint Service.
+///
+/// A Service is created by the factory during server initialization.
 ///
-/// Transform factory creates service that wraps other services.
 /// `Config` is a service factory configuration type.
 pub trait Transform<S> {
     /// Requests handled by the service.
@@ -33,7 +40,7 @@ pub trait Transform<S> {
     /// The future response value.
     type Future: Future<Item = Self::Transform, Error = Self::InitError>;
 
-    /// Create and return a new service value asynchronously.
+    /// Creates and returns a new Service component, asynchronously
     fn new_transform(&self, service: S) -> Self::Future;
 
     /// Map this service's factory error to a different error,

actix-test-server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-test-server"
-version = "0.2.0"
+version = "0.2.1"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix test server"
 keywords = ["network", "framework", "async", "futures"]

actix-test-server/src/lib.rs

@@ -6,7 +6,7 @@ use actix_rt::{Runtime, System};
 use actix_server::{Server, StreamServiceFactory};
 pub use actix_server_config::{Io, ServerConfig};
 
-use futures::Future;
+use futures::future::{lazy, Future, IntoFuture};
 use net2::TcpBuilder;
 use tokio_reactor::Handle;
 use tokio_tcp::TcpStream;
@@ -99,6 +99,15 @@ impl TestServerRuntime {
         self.rt.block_on(fut)
     }
 
+    /// Runs the provided function, with runtime enabled.
+    pub fn run_on<F, R>(&mut self, f: F) -> Result<R::Item, R::Error>
+    where
+        F: FnOnce() -> R,
+        R: IntoFuture,
+    {
+        self.rt.block_on(lazy(|| f().into_future()))
+    }
+
     /// Spawn future to the current runtime
     pub fn spawn<F>(&mut self, fut: F)
     where

actix-threadpool/CHANGES.md

@@ -0,0 +1,5 @@
+# Changes
+
+## [0.1.0] - 2019-03-28
+
+* Move threadpool to separate crate

actix-threadpool/Cargo.toml

@@ -0,0 +1,27 @@
+[package]
+name = "actix-threadpool"
+version = "0.1.0"
+authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
+description = "Actix thread pool for sync code"
+keywords = ["actix", "network", "framework", "async", "futures"]
+homepage = "https://actix.rs"
+repository = "https://github.com/actix/actix-net.git"
+documentation = "https://docs.rs/actix-threadpool/"
+categories = ["network-programming", "asynchronous"]
+license = "MIT/Apache-2.0"
+exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
+edition = "2018"
+workspace = ".."
+
+[lib]
+name = "actix_threadpool"
+path = "src/lib.rs"
+
+[dependencies]
+derive_more = "0.14"
+futures = "0.1.25"
+parking_lot = "0.7"
+lazy_static = "1.2"
+log = "0.4"
+num_cpus = "1.10"
+threadpool = "1.7"

actix-threadpool/src/lib.rs

@@ -9,7 +9,7 @@ use parking_lot::Mutex;
 use threadpool::ThreadPool;
 
 /// Env variable for default cpu pool size
-const ENV_CPU_POOL_VAR: &str = "ACTIX_CPU_POOL";
+const ENV_CPU_POOL_VAR: &str = "ACTIX_THREADPOOL";
 
 lazy_static::lazy_static! {
     pub(crate) static ref DEFAULT_POOL: Mutex<ThreadPool> = {
@@ -18,7 +18,7 @@ lazy_static::lazy_static! {
                 if let Ok(val) = val.parse() {
                     val
                 } else {
-                    log::error!("Can not parse ACTIX_CPU_POOL value");
+                    log::error!("Can not parse ACTIX_THREADPOOL value");
                     num_cpus::get() * 5
                 }
             }

actix-utils/CHANGES.md

@@ -1,5 +1,17 @@
 # Changes
 
+
+## [0.3.5] - 2019-04-04
+
+### Added
+
+* Allow to send messages to `FramedTransport` via mpsc channel.
+
+### Changed
+
+* Remove 'static constraint from Clonable service
+
+
 ## [0.3.4] - 2019-03-12
 
 ### Changed

actix-utils/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-utils"
-version = "0.3.4"
+version = "0.3.5"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix utils - various actix net related services"
 keywords = ["network", "framework", "async", "futures"]

actix-utils/src/cloneable.rs

@@ -7,12 +7,12 @@ use futures::Poll;
 use super::cell::Cell;
 
 /// Service that allows to turn non-clone service to a service with `Clone` impl
-pub struct CloneableService<T: 'static> {
+pub struct CloneableService<T> {
     service: Cell<T>,
     _t: PhantomData<Rc<()>>,
 }
 
-impl<T: 'static> CloneableService<T> {
+impl<T> CloneableService<T> {
     pub fn new(service: T) -> Self
     where
         T: Service,
@@ -24,7 +24,7 @@ impl<T: 'static> CloneableService<T> {
     }
 }
 
-impl<T: 'static> Clone for CloneableService<T> {
+impl<T> Clone for CloneableService<T> {
     fn clone(&self) -> Self {
         Self {
             service: self.service.clone(),
@@ -35,7 +35,7 @@ impl<T: 'static> Clone for CloneableService<T> {
 
 impl<T> Service for CloneableService<T>
 where
-    T: Service + 'static,
+    T: Service,
 {
     type Request = T::Request;
     type Response = T::Response;

actix-utils/src/framed.rs

@@ -7,6 +7,7 @@ use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
 use actix_service::{IntoNewService, IntoService, NewService, Service};
 use futures::future::{ok, FutureResult};
 use futures::task::AtomicTask;
+use futures::unsync::mpsc;
 use futures::{Async, Future, IntoFuture, Poll, Sink, Stream};
 use log::debug;
 
@@ -178,6 +179,11 @@ impl<E, U: Encoder + Decoder> From<E> for FramedTransportError<E, U> {
     }
 }
 
+pub enum FramedMessage<T> {
+    Message(T),
+    Close,
+}
+
 /// FramedTransport - is a future that reads frames from Framed object
 /// and pass then to the service.
 pub struct FramedTransport<S, T, U>
@@ -193,6 +199,7 @@ where
     service: S,
     state: TransportState<S, U>,
     framed: Framed<T, U>,
+    rx: Option<mpsc::UnboundedReceiver<FramedMessage<<U as Encoder>::Item>>>,
     inner: Cell<FramedTransportInner<<U as Encoder>::Item, S::Error>>,
 }
 
@@ -200,6 +207,7 @@ enum TransportState<S: Service, U: Encoder + Decoder> {
     Processing,
     Error(FramedTransportError<S::Error, U>),
     FramedError(FramedTransportError<S::Error, U>),
+    FlushAndStop,
     Stopping,
 }
 
@@ -257,10 +265,12 @@ where
     /// write to framed object
     fn poll_write(&mut self) -> bool {
         let inner = self.inner.get_mut();
+        let mut rx_done = self.rx.is_none();
+        let mut buf_empty = inner.buf.is_empty();
         loop {
             while !self.framed.is_write_buf_full() {
-                if let Some(msg) = inner.buf.pop_front() {
-                    match msg {
+                if !buf_empty {
+                    match inner.buf.pop_front().unwrap() {
                         Ok(msg) => {
                             if let Err(err) = self.framed.force_send(msg) {
                                 self.state = TransportState::FramedError(
@@ -268,6 +278,7 @@ where
                                 );
                                 return true;
                             }
+                            buf_empty = inner.buf.is_empty();
                         }
                         Err(err) => {
                             self.state =
@@ -275,7 +286,33 @@ where
                             return true;
                         }
                     }
-                } else {
+                }
+
+                if !rx_done && self.rx.is_some() {
+                    match self.rx.as_mut().unwrap().poll() {
+                        Ok(Async::Ready(Some(FramedMessage::Message(msg)))) => {
+                            if let Err(err) = self.framed.force_send(msg) {
+                                self.state = TransportState::FramedError(
+                                    FramedTransportError::Encoder(err),
+                                );
+                                return true;
+                            }
+                        }
+                        Ok(Async::Ready(Some(FramedMessage::Close))) => {
+                            self.state = TransportState::FlushAndStop;
+                            return true;
+                        }
+                        Ok(Async::Ready(None)) => {
+                            let _ = self.rx.take();
+                        }
+                        Ok(Async::NotReady) => rx_done = true,
+                        Err(_e) => {
+                            let _ = self.rx.take();
+                        }
+                    }
+                }
+
+                if rx_done && buf_empty {
                     break;
                 }
             }
@@ -313,6 +350,7 @@ where
     pub fn new<F: IntoService<S>>(framed: Framed<T, U>, service: F) -> Self {
         FramedTransport {
             framed,
+            rx: None,
             service: service.into_service(),
             state: TransportState::Processing,
             inner: Cell::new(FramedTransportInner {
@@ -322,6 +360,15 @@ where
         }
     }
 
+    /// Get Sender
+    pub fn set_receiver(
+        mut self,
+        rx: mpsc::UnboundedReceiver<FramedMessage<<U as Encoder>::Item>>,
+    ) -> Self {
+        self.rx = Some(rx);
+        self
+    }
+
     /// Get reference to a service wrapped by `FramedTransport` instance.
     pub fn get_ref(&self) -> &S {
         &self.service
@@ -378,6 +425,20 @@ where
                     Ok(Async::NotReady)
                 }
             }
+            TransportState::FlushAndStop => {
+                if !self.framed.is_write_buf_empty() {
+                    match self.framed.poll_complete() {
+                        Err(err) => {
+                            debug!("Error sending data: {:?}", err);
+                            Ok(Async::Ready(()))
+                        }
+                        Ok(Async::NotReady) => Ok(Async::NotReady),
+                        Ok(Async::Ready(_)) => Ok(Async::Ready(())),
+                    }
+                } else {
+                    Ok(Async::Ready(()))
+                }
+            }
             TransportState::FramedError(err) => Err(err),
             TransportState::Stopping => Ok(Async::Ready(())),
         }

router/CHANGES.txt

@@ -1,5 +1,15 @@
 # Changes
 
+## [0.1.2] - 2019-04-07
+
+* Export `Quoter` type
+
+* Allow to reset `Path` instance
+
+## [0.1.1] - 2019-04-03
+
+* Get dynamic segment by name instead of iterator.
+
 ## [0.1.0] - 2019-03-09
 
 * Initial release

router/Cargo.toml

@@ -1,16 +1,16 @@
 [package]
 name = "actix-router"
-version = "0.1.0"
+version = "0.1.2"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Path router"
 keywords = ["actix"]
 homepage = "https://actix.rs"
 repository = "https://github.com/actix/actix-net.git"
-documentation = "https://actix.rs/api/actix-net/stable/actix_router/"
+documentation = "https://docs.rs/actix-router/"
 license = "MIT/Apache-2.0"
 exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
 edition = "2018"
-workspace = "../"
+workspace = ".."
 
 [lib]
 name = "actix_router"
@@ -23,7 +23,8 @@ default = ["http"]
 bytes = "0.4"
 regex = "1.0"
 serde = "1.0.80"
-string = "0.1.3"
+string = "0.2.0"
+log = "0.4"
 http = { version="0.1.14", optional=true }
 
 [dev-dependencies]

router/src/lib.rs

@@ -39,7 +39,7 @@ impl<T: AsRef<[u8]>> ResourcePath for string::String<T> {
 mod url;
 
 #[cfg(feature = "http")]
-pub use self::url::Url;
+pub use self::url::{Quoter, Url};
 
 #[cfg(feature = "http")]
 mod http_support {

router/src/path.rs

@@ -51,16 +51,19 @@ impl<T: ResourcePath> Path<T> {
         }
     }
 
+    #[inline]
     /// Get reference to inner path instance
     pub fn get_ref(&self) -> &T {
         &self.path
     }
 
+    #[inline]
     /// Get mutable reference to inner path instance
     pub fn get_mut(&mut self) -> &mut T {
         &mut self.path
     }
 
+    #[inline]
     /// Path
     pub fn path(&self) -> &str {
         let skip = self.skip as usize;
@@ -72,13 +75,22 @@ impl<T: ResourcePath> Path<T> {
         }
     }
 
-    /// Reset inner path
+    #[inline]
+    /// Set new path
     pub fn set(&mut self, path: T) {
         self.skip = 0;
         self.path = path;
         self.segments.clear();
     }
 
+    #[inline]
+    /// Reset state
+    pub fn reset(&mut self) {
+        self.skip = 0;
+        self.segments.clear();
+    }
+
+    #[inline]
     /// Skip first `n` chars in path
     pub fn skip(&mut self, n: u16) {
         self.skip = self.skip + n;
@@ -99,11 +111,13 @@ impl<T: ResourcePath> Path<T> {
             .push((Rc::new(name.to_string()), PathItem::Static(value)));
     }
 
+    #[inline]
     /// Check if there are any matched patterns
     pub fn is_empty(&self) -> bool {
         self.segments.is_empty()
     }
 
+    #[inline]
     /// Check number of extracted parameters
     pub fn len(&self) -> usize {
         self.segments.len()

router/src/resource.rs

@@ -118,6 +118,7 @@ impl ResourceDef {
         &self.pattern
     }
 
+    #[inline]
     /// Check if path matchs this pattern?
     pub fn is_match(&self, path: &str) -> bool {
         match self.tp {
@@ -196,18 +197,17 @@ impl ResourceDef {
                     [PathItem::Static(""); MAX_DYNAMIC_SEGMENTS];
 
                 if let Some(captures) = re.captures(path.path()) {
-                    let mut passed = false;
-
-                    for capture in captures.iter() {
-                        if let Some(ref m) = capture {
-                            if !passed {
-                                passed = true;
-                                continue;
-                            }
-
-                            segments[idx] = PathItem::Segment(m.start() as u16, m.end() as u16);
+                    for (no, name) in names.iter().enumerate() {
+                        if let Some(m) = captures.name(&name) {
                             idx += 1;
                             pos = m.end();
+                            segments[no] = PathItem::Segment(m.start() as u16, m.end() as u16);
+                        } else {
+                            log::error!(
+                                "Dynamic path match but not all segments found: {}",
+                                name
+                            );
+                            false;
                         }
                     }
                 } else {

router/src/url.rs

@@ -1,5 +1,3 @@
-use std::rc::Rc;
-
 use crate::ResourcePath;
 
 #[allow(dead_code)]
@@ -39,7 +37,7 @@ thread_local! {
 #[derive(Default, Clone, Debug)]
 pub struct Url {
     uri: http::Uri,
-    path: Option<Rc<String>>,
+    path: Option<String>,
 }
 
 impl Url {
@@ -49,6 +47,13 @@ impl Url {
         Url { uri, path }
     }
 
+    pub fn with_quoter(uri: http::Uri, quoter: &Quoter) -> Url {
+        Url {
+            path: quoter.requote(uri.path().as_bytes()),
+            uri,
+        }
+    }
+
     pub fn uri(&self) -> &http::Uri {
         &self.uri
     }
@@ -61,19 +66,27 @@ impl Url {
         }
     }
 
+    #[inline]
     pub fn update(&mut self, uri: &http::Uri) {
         self.uri = uri.clone();
         self.path = DEFAULT_QUOTER.with(|q| q.requote(uri.path().as_bytes()));
     }
+
+    #[inline]
+    pub fn update_with_quoter(&mut self, uri: &http::Uri, quoter: &Quoter) {
+        self.uri = uri.clone();
+        self.path = quoter.requote(uri.path().as_bytes());
+    }
 }
 
 impl ResourcePath for Url {
+    #[inline]
     fn path(&self) -> &str {
         self.path()
     }
 }
 
-pub(crate) struct Quoter {
+pub struct Quoter {
     safe_table: [u8; 16],
     protected_table: [u8; 16],
 }
@@ -108,7 +121,7 @@ impl Quoter {
         q
     }
 
-    pub fn requote(&self, val: &[u8]) -> Option<Rc<String>> {
+    pub fn requote(&self, val: &[u8]) -> Option<String> {
         let mut has_pct = 0;
         let mut pct = [b'%', 0, 0];
         let mut idx = 0;
@@ -160,7 +173,7 @@ impl Quoter {
         if let Some(data) = cloned {
             // Unsafe: we get data from http::Uri, which does utf-8 checks already
             // this code only decodes valid pct encoded values
-            Some(Rc::new(unsafe { String::from_utf8_unchecked(data) }))
+            Some(unsafe { String::from_utf8_unchecked(data) })
         } else {
             None
         }