diff --git a/actix-server/Cargo.toml b/actix-server/Cargo.toml
index 011c76c3..ca9d0a6e 100644
--- a/actix-server/Cargo.toml
+++ b/actix-server/Cargo.toml
@@ -45,12 +45,13 @@ num_cpus = "1.0"
mio = "0.6.19"
net2 = "0.2"
-futures = { package = "futures-preview", version = "0.3.0-alpha.18" }
+futures = "0.3.1"
slab = "0.4"
-tokio = "0.2.0-alpha.4"
-tokio-io = "0.2.0-alpha.4"
-tokio-net = { version = "0.2.0-alpha.4", features = ["signal"] }
-tokio-timer = "0.3.0-alpha.4"
+
+tokio = "0.2.0-alpha.6"
+tokio-io = "0.2.0-alpha.6"
+tokio-net = { version = "0.2.0-alpha.6", features = ["signal"] }
+tokio-timer = "0.3.0-alpha.6"
# unix domain sockets
mio-uds = { version="0.6.7", optional = true }
@@ -61,7 +62,7 @@ native-tls = { version="0.2", optional = true }
# openssl
openssl = { version="0.10", optional = true }
-tokio-openssl = { version="0.4.0-alpha.4", optional = true }
+tokio-openssl = { version="0.4.0-alpha.6", optional = true }
# rustls
rustls = { version = "0.16.0", optional = true }
diff --git a/actix-server/src/accept.rs b/actix-server/src/accept.rs
index 597471ea..421aa0ab 100644
--- a/actix-server/src/accept.rs
+++ b/actix-server/src/accept.rs
@@ -3,7 +3,7 @@ use std::time::{Duration, Instant};
use std::{io, thread};
use actix_rt::System;
-
+use futures::FutureExt;
use log::{error, info};
use slab::Slab;
use tokio_timer::delay;
@@ -12,7 +12,6 @@ use crate::server::Server;
use crate::socket::{SocketAddr, SocketListener, StdListener};
use crate::worker::{Conn, WorkerClient};
use crate::Token;
-use futures::{Future, FutureExt};
pub(crate) enum Command {
Pause,
@@ -371,7 +370,7 @@ impl Accept {
match self.workers[self.next].send(msg) {
Ok(_) => (),
Err(tmp) => {
- self.srv.worker_died(self.workers[self.next].idx);
+ self.srv.worker_faulted(self.workers[self.next].idx);
msg = tmp;
self.workers.swap_remove(self.next);
if self.workers.is_empty() {
@@ -397,7 +396,7 @@ impl Accept {
return;
}
Err(tmp) => {
- self.srv.worker_died(self.workers[self.next].idx);
+ self.srv.worker_faulted(self.workers[self.next].idx);
msg = tmp;
self.workers.swap_remove(self.next);
if self.workers.is_empty() {
@@ -444,7 +443,7 @@ impl Accept {
System::current().arbiter().send(
async move {
delay(Instant::now() + Duration::from_millis(510)).await;
- r.set_readiness(mio::Ready::readable());
+ let _ = r.set_readiness(mio::Ready::readable());
}
.boxed(),
);
diff --git a/actix-server/src/builder.rs b/actix-server/src/builder.rs
index 7a22b156..6666f8fd 100644
--- a/actix-server/src/builder.rs
+++ b/actix-server/src/builder.rs
@@ -1,27 +1,27 @@
-use std::time::Duration;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use std::time::{Duration, Instant};
use std::{io, mem, net};
use actix_rt::{spawn, Arbiter, System};
use futures::channel::mpsc::{unbounded, UnboundedReceiver};
use futures::future::ready;
use futures::stream::FuturesUnordered;
-use futures::{ready, Future, FutureExt, Poll, Stream, StreamExt};
+use futures::{ready, Future, FutureExt, Stream, StreamExt};
use log::{error, info};
use net2::TcpBuilder;
use num_cpus;
use tokio_net::tcp::TcpStream;
-use tokio_timer::sleep;
+use tokio_timer::delay;
use crate::accept::{AcceptLoop, AcceptNotify, Command};
use crate::config::{ConfiguredService, ServiceConfig};
use crate::server::{Server, ServerCommand};
use crate::services::{InternalServiceFactory, ServiceFactory, StreamNewService};
-use crate::signals::{Signal, Signals};
+// use crate::signals::{Signal, Signals};
use crate::socket::StdListener;
use crate::worker::{self, Worker, WorkerAvailability, WorkerClient};
use crate::{ssl, Token};
-use std::pin::Pin;
-use std::task::Context;
/// Server builder
pub struct ServerBuilder {
@@ -303,7 +303,7 @@ impl ServerBuilder {
// handle signals
if !self.no_signals {
- Signals::start(self.server.clone());
+ // Signals::start(self.server.clone());
}
// start http server actor
@@ -342,37 +342,37 @@ impl ServerBuilder {
self.accept.send(Command::Resume);
let _ = tx.send(());
}
- ServerCommand::Signal(sig) => {
- // Signals support
- // Handle `SIGINT`, `SIGTERM`, `SIGQUIT` signals and stop actix system
- match sig {
- Signal::Int => {
- info!("SIGINT received, exiting");
- self.exit = true;
- self.handle_cmd(ServerCommand::Stop {
- graceful: false,
- completion: None,
- })
- }
- Signal::Term => {
- info!("SIGTERM received, stopping");
- self.exit = true;
- self.handle_cmd(ServerCommand::Stop {
- graceful: true,
- completion: None,
- })
- }
- Signal::Quit => {
- info!("SIGQUIT received, exiting");
- self.exit = true;
- self.handle_cmd(ServerCommand::Stop {
- graceful: false,
- completion: None,
- })
- }
- _ => (),
- }
- }
+ // ServerCommand::Signal(sig) => {
+ // Signals support
+ // Handle `SIGINT`, `SIGTERM`, `SIGQUIT` signals and stop actix system
+ // match sig {
+ // Signal::Int => {
+ // info!("SIGINT received, exiting");
+ // self.exit = true;
+ // self.handle_cmd(ServerCommand::Stop {
+ // graceful: false,
+ // completion: None,
+ // })
+ // }
+ // Signal::Term => {
+ // info!("SIGTERM received, stopping");
+ // self.exit = true;
+ // self.handle_cmd(ServerCommand::Stop {
+ // graceful: true,
+ // completion: None,
+ // })
+ // }
+ // Signal::Quit => {
+ // info!("SIGQUIT received, exiting");
+ // self.exit = true;
+ // self.handle_cmd(ServerCommand::Stop {
+ // graceful: false,
+ // completion: None,
+ // })
+ // }
+ // _ => (),
+ // }
+ // }
ServerCommand::Stop {
graceful,
completion,
@@ -397,7 +397,7 @@ impl ServerBuilder {
if exit {
spawn(
async {
- tokio_timer::sleep(Duration::from_millis(300))
+ delay(Instant::now() + Duration::from_millis(300))
.await;
System::current().stop();
}
@@ -410,17 +410,19 @@ impl ServerBuilder {
} else {
// we need to stop system if server was spawned
if self.exit {
- spawn(sleep(Duration::from_millis(300)).then(|_| {
- System::current().stop();
- ready(())
- }));
+ spawn(
+ delay(Instant::now() + Duration::from_millis(300)).then(|_| {
+ System::current().stop();
+ ready(())
+ }),
+ );
}
if let Some(tx) = completion {
let _ = tx.send(());
}
}
}
- ServerCommand::WorkerDied(idx) => {
+ ServerCommand::WorkerFaulted(idx) => {
let mut found = false;
for i in 0..self.workers.len() {
if self.workers[i].0 == idx {
diff --git a/actix-server/src/config.rs b/actix-server/src/config.rs
index e8b84d56..35176841 100644
--- a/actix-server/src/config.rs
+++ b/actix-server/src/config.rs
@@ -1,12 +1,10 @@
use std::collections::HashMap;
use std::{fmt, io, net};
-use crate::counter::CounterGuard;
use actix_server_config::{Io, ServerConfig};
-use actix_service::{IntoNewService, NewService, ServiceExt};
-use futures::future::{join_all, Future, FutureExt, LocalBoxFuture, TryFutureExt};
+use actix_service::{Factory, IntoFactory};
+use futures::future::{Future, FutureExt, LocalBoxFuture};
use log::error;
-use std::pin::Pin;
use tokio_net::tcp::TcpStream;
use super::builder::bind_addr;
@@ -14,7 +12,7 @@ use super::services::{
BoxedServerService, InternalServiceFactory, ServerMessage, StreamService,
};
use super::Token;
-use std::process::Output;
+use crate::counter::CounterGuard;
pub struct ServiceConfig {
pub(crate) services: Vec<(String, net::TcpListener)>,
@@ -115,7 +113,7 @@ impl InternalServiceFactory for ConfiguredService {
self.rt.configure(&mut rt);
rt.validate();
- let mut names = self.names.clone();
+ let names = self.names.clone();
// construct services
async move {
@@ -197,23 +195,20 @@ impl ServiceRuntime {
/// *ServiceConfig::bind()* or *ServiceConfig::listen()* methods.
pub fn service<T, F>(&mut self, name: &str, service: F)
where
- F: IntoNewService<T>,
- T: NewService<Config = ServerConfig, Request = Io<TcpStream>> + 'static,
+ F: IntoFactory<T>,
+ T: Factory<Config = ServerConfig, Request = Io<TcpStream>> + '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(),
- }),
- );
-
+ token.clone(),
+ Box::new(ServiceFactory {
+ inner: service.into_factory(),
+ }),
+ );
} else {
panic!("Unknown service: {:?}", name);
}
@@ -229,7 +224,7 @@ impl ServiceRuntime {
}
type BoxedNewService = Box<
- dyn NewService<
+ dyn Factory<
Request = (Option<CounterGuard>, ServerMessage),
Response = (),
Error = (),
@@ -244,9 +239,9 @@ struct ServiceFactory<T> {
inner: T,
}
-impl<T> NewService for ServiceFactory<T>
+impl<T> Factory for ServiceFactory<T>
where
- T: NewService<Config = ServerConfig, Request = Io<TcpStream>>,
+ T: Factory<Config = ServerConfig, Request = Io<TcpStream>>,
T::Future: 'static,
T::Service: 'static,
T::Error: 'static,
@@ -260,14 +255,15 @@ where
type Service = BoxedServerService;
type Future = LocalBoxFuture<'static, Result<BoxedServerService, ()>>;
- // Box<dyn Future<Output=Result<Vec<(Token, BoxedServerService)>, ()>>>;
-
fn new_service(&self, cfg: &ServerConfig) -> Self::Future {
let fut = self.inner.new_service(cfg);
async move {
return match fut.await {
Ok(s) => Ok(Box::new(StreamService::new(s)) as BoxedServerService),
- Err(e) => Err(()),
+ Err(e) => {
+ error!("Can not construct service: {:?}", e);
+ Err(())
+ }
};
}
.boxed_local()
diff --git a/actix-server/src/lib.rs b/actix-server/src/lib.rs
index 3e8cc8fc..24390859 100644
--- a/actix-server/src/lib.rs
+++ b/actix-server/src/lib.rs
@@ -6,7 +6,7 @@ mod config;
mod counter;
mod server;
mod services;
-mod signals;
+// mod signals;
mod socket;
pub mod ssl;
mod worker;
diff --git a/actix-server/src/server.rs b/actix-server/src/server.rs
index bb7f0649..67f9ca5d 100644
--- a/actix-server/src/server.rs
+++ b/actix-server/src/server.rs
@@ -3,14 +3,14 @@ use futures::channel::oneshot;
use futures::{Future, TryFutureExt};
use crate::builder::ServerBuilder;
-use crate::signals::Signal;
+// use crate::signals::Signal;
#[derive(Debug)]
pub(crate) enum ServerCommand {
- WorkerDied(usize),
+ WorkerFaulted(usize),
Pause(oneshot::Sender<()>),
Resume(oneshot::Sender<()>),
- Signal(Signal),
+ // Signal(Signal),
/// Whether to try and shut down gracefully
Stop {
graceful: bool,
@@ -31,12 +31,12 @@ impl Server {
ServerBuilder::default()
}
- pub(crate) fn signal(&self, sig: Signal) {
- let _ = self.0.unbounded_send(ServerCommand::Signal(sig));
- }
+ // pub(crate) fn signal(&self, sig: Signal) {
+ // let _ = self.0.unbounded_send(ServerCommand::Signal(sig));
+ // }
- pub(crate) fn worker_died(&self, idx: usize) {
- let _ = self.0.unbounded_send(ServerCommand::WorkerDied(idx));
+ pub(crate) fn worker_faulted(&self, idx: usize) {
+ let _ = self.0.unbounded_send(ServerCommand::WorkerFaulted(idx));
}
/// Pause accepting incoming connections
diff --git a/actix-server/src/services.rs b/actix-server/src/services.rs
index 5d6181a5..d0bffadf 100644
--- a/actix-server/src/services.rs
+++ b/actix-server/src/services.rs
@@ -1,19 +1,18 @@
use std::marker::PhantomData;
use std::net::SocketAddr;
+use std::task::{Context, Poll};
use std::time::Duration;
use actix_rt::spawn;
use actix_server_config::{Io, ServerConfig};
-use actix_service::{NewService, Service, ServiceExt};
+use actix_service::{Factory, Service};
use futures::future::{err, ok, LocalBoxFuture, Ready};
-use futures::{Future, FutureExt, Poll, StreamExt, TryFutureExt};
+use futures::{FutureExt, TryFutureExt};
use log::error;
use super::Token;
use crate::counter::CounterGuard;
use crate::socket::{FromStream, StdStream};
-use std::pin::Pin;
-use std::task::Context;
/// Server message
pub(crate) enum ServerMessage {
@@ -26,7 +25,7 @@ pub(crate) enum ServerMessage {
}
pub trait ServiceFactory<Stream: FromStream>: Send + Clone + 'static {
- type NewService: NewService<Config = ServerConfig, Request = Io<Stream>>;
+ type NewService: Factory<Config = ServerConfig, Request = Io<Stream>>;
fn create(&self) -> Self::NewService;
}
@@ -70,18 +69,10 @@ where
type Error = ();
type Future = Ready<Result<(), ()>>;
- fn poll_ready(
- self: Pin<&mut Self>,
- ctx: &mut Context<'_>,
- ) -> Poll<Result<(), Self::Error>> {
- unimplemented!()
+ fn poll_ready(&mut self, ctx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+ self.service.poll_ready(ctx).map_err(|_| ())
}
- /*
- fn poll_ready(&mut self) -> Poll<(), Self::Error> {
- self.service.poll_ready().map_err(|_| ())
- }
- */
fn call(&mut self, (guard, req): (Option<CounterGuard>, ServerMessage)) -> Self::Future {
match req {
ServerMessage::Connect(stream) => {
@@ -93,7 +84,7 @@ where
let f = self.service.call(Io::new(stream));
spawn(
async move {
- f.await;
+ let _ = f.await;
drop(guard);
}
.boxed_local(),
@@ -189,7 +180,7 @@ impl InternalServiceFactory for Box<dyn InternalServiceFactory> {
impl<F, T, I> ServiceFactory<I> for F
where
F: Fn() -> T + Send + Clone + 'static,
- T: NewService<Config = ServerConfig, Request = Io<I>>,
+ T: Factory<Config = ServerConfig, Request = Io<I>>,
I: FromStream,
{
type NewService = T;
diff --git a/actix-server/src/signals.rs b/actix-server/src/signals.rs
index 85c88734..f87cf18b 100644
--- a/actix-server/src/signals.rs
+++ b/actix-server/src/signals.rs
@@ -1,17 +1,15 @@
+use std::future::Future;
use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
use actix_rt::spawn;
+use futures::future::LocalBoxFuture;
use futures::stream::{futures_unordered, FuturesUnordered, LocalBoxStream};
-use futures::{
- Future, FutureExt, Poll, Stream, StreamExt, TryFutureExt, TryStream, TryStreamExt,
-};
+use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStream, TryStreamExt};
+use tokio_net::signal::unix::signal;
use crate::server::Server;
-use actix_service::ServiceExt;
-use futures::future::LocalBoxFuture;
-use std::pin::Pin;
-use std::task::Context;
-use tokio_net::signal::unix::signal;
/// Different types of process signals
#[derive(PartialEq, Clone, Copy, Debug)]
diff --git a/actix-server/src/worker.rs b/actix-server/src/worker.rs
index 613d8eaa..94f2c3c2 100644
--- a/actix-server/src/worker.rs
+++ b/actix-server/src/worker.rs
@@ -1,23 +1,22 @@
+use std::pin::Pin;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::Arc;
-use std::{mem, task, time};
+use std::task::{Context, Poll};
+use std::{mem, time};
+use actix_rt::{spawn, Arbiter};
use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender};
use futures::channel::oneshot;
-use futures::{future, Future, Poll, Stream, TryFutureExt};
-use futures::{FutureExt, StreamExt};
+use futures::future::{join_all, LocalBoxFuture, MapOk};
+use futures::{Future, FutureExt, Stream, TryFutureExt};
use log::{error, info, trace};
-use tokio_timer::{sleep, Delay};
+use tokio_timer::{delay, Delay};
use crate::accept::AcceptNotify;
use crate::counter::Counter;
use crate::services::{BoxedServerService, InternalServiceFactory, ServerMessage};
use crate::socket::{SocketAddr, StdStream};
use crate::Token;
-use actix_rt::spawn;
-use futures::future::{LocalBoxFuture, MapOk};
-use std::pin::Pin;
-use std::task::Context;
pub(crate) struct WorkerCommand(Conn);
@@ -167,8 +166,8 @@ impl Worker {
}
spawn(
- async {
- let mut res = future::join_all(fut).await;
+ async move {
+ let res = join_all(fut).await;
let res: Result<Vec<_>, _> = res.into_iter().collect();
match res {
Ok(services) => {
@@ -177,12 +176,13 @@ impl Worker {
while token.0 >= wrk.services.len() {
wrk.services.push(None);
}
+ wrk.services[token.0] = Some((idx, service));
}
}
- Ok::<_, ()>(wrk);
}
Err(e) => {
- //return Err(e);
+ error!("Can not start worker: {:?}", e);
+ Arbiter::current().stop();
}
}
}
@@ -212,8 +212,7 @@ impl Worker {
trace: bool,
cx: &mut Context<'_>,
) -> Result<bool, (Token, usize)> {
- /*
- let mut ready = self.conns.available();
+ let mut ready = self.conns.available(cx);
let mut failed = None;
for (token, service) in &mut self.services.iter_mut().enumerate() {
if let Some(service) = service {
@@ -226,7 +225,7 @@ impl Worker {
);
}
}
- Poll::NotReady => ready = false,
+ Poll::Pending => ready = false,
Poll::Ready(Err(_)) => {
error!(
"Service {:?} readiness check returned error, restarting",
@@ -241,8 +240,7 @@ impl Worker {
Err(idx)
} else {
Ok(ready)
- }*/
- unimplemented!()
+ }
}
}
@@ -253,7 +251,7 @@ enum WorkerState {
Restarting(
usize,
Token,
- Box<dyn Future<Output = Result<Vec<(Token, BoxedServerService)>, ()>>>,
+ Pin<Box<dyn Future<Output = Result<Vec<(Token, BoxedServerService)>, ()>>>>,
),
Shutdown(Delay, Delay, oneshot::Sender<bool>),
}
@@ -261,39 +259,36 @@ enum WorkerState {
impl Future for Worker {
type Output = ();
- fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
- unimplemented!()
- }
-
- /*
- fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+ fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
// `StopWorker` message handler
- if let Ok(Async::Ready(Some(StopCommand { graceful, result }))) = self.rx2.poll() {
+ if let Poll::Ready(Some(StopCommand { graceful, result })) =
+ Pin::new(&mut self.rx2).poll_next(cx)
+ {
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(()));
+ return Poll::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),
+ delay(time::Instant::now() + time::Duration::from_secs(1)),
+ delay(time::Instant::now() + self.shutdown_timeout),
result,
);
} else {
let _ = result.send(true);
- return Ok(Async::Ready(()));
+ return Poll::Ready(());
}
} else {
info!("Force shutdown worker, {} connections", num);
self.shutdown(true);
let _ = result.send(false);
- return Ok(Async::Ready(()));
+ return Poll::Ready(());
}
}
@@ -301,13 +296,13 @@ impl Future for Worker {
match state {
WorkerState::Unavailable(mut conns) => {
- match self.check_readiness(true) {
+ match self.check_readiness(true, cx) {
Ok(true) => {
self.state = WorkerState::Available;
// process requests from wait queue
while let Some(msg) = conns.pop() {
- match self.check_readiness(false) {
+ match self.check_readiness(false, cx) {
Ok(true) => {
let guard = self.conns.get();
let _ = self.services[msg.token.0]
@@ -319,7 +314,7 @@ impl Future for Worker {
Ok(false) => {
trace!("Worker is unavailable");
self.state = WorkerState::Unavailable(conns);
- return self.poll();
+ return self.poll(cx);
}
Err((token, idx)) => {
trace!(
@@ -331,16 +326,16 @@ impl Future for Worker {
token,
self.factories[idx].create(),
);
- return self.poll();
+ return self.poll(cx);
}
}
}
self.availability.set(true);
- return self.poll();
+ return self.poll(cx);
}
Ok(false) => {
self.state = WorkerState::Unavailable(conns);
- return Ok(Async::NotReady);
+ return Poll::Pending;
}
Err((token, idx)) => {
trace!(
@@ -349,13 +344,13 @@ impl Future for Worker {
);
self.state =
WorkerState::Restarting(idx, token, self.factories[idx].create());
- return self.poll();
+ return self.poll(cx);
}
}
}
WorkerState::Restarting(idx, token, mut fut) => {
- match fut.poll() {
- Ok(Async::Ready(item)) => {
+ match Pin::new(&mut fut).poll(cx) {
+ Poll::Ready(Ok(item)) => {
for (token, service) in item {
trace!(
"Service {:?} has been restarted",
@@ -365,55 +360,55 @@ impl Future for Worker {
self.state = WorkerState::Unavailable(Vec::new());
}
}
- Ok(Async::NotReady) => {
- self.state = WorkerState::Restarting(idx, token, fut);
- return Ok(Async::NotReady);
- }
- Err(_) => {
+ Poll::Ready(Err(_)) => {
panic!(
"Can not restart {:?} service",
self.factories[idx].name(token)
);
}
+ Poll::Pending => {
+ self.state = WorkerState::Restarting(idx, token, fut);
+ return Poll::Pending;
+ }
}
- return self.poll();
+ return self.poll(cx);
}
WorkerState::Shutdown(mut t1, mut t2, tx) => {
let num = num_connections();
if num == 0 {
let _ = tx.send(true);
Arbiter::current().stop();
- return Ok(Async::Ready(()));
+ return Poll::Ready(());
}
// check graceful timeout
- match t2.poll().unwrap() {
- Async::NotReady => (),
- Async::Ready(_) => {
+ match Pin::new(&mut t2).poll(cx) {
+ Poll::Pending => (),
+ Poll::Ready(_) => {
self.shutdown(true);
let _ = tx.send(false);
Arbiter::current().stop();
- return Ok(Async::Ready(()));
+ return Poll::Ready(());
}
}
// sleep for 1 second and then check again
- match t1.poll().unwrap() {
- Async::NotReady => (),
- Async::Ready(_) => {
- t1 = sleep(time::Duration::from_secs(1));
- let _ = t1.poll();
+ match Pin::new(&mut t1).poll(cx) {
+ Poll::Pending => (),
+ Poll::Ready(_) => {
+ t1 = delay(time::Instant::now() + time::Duration::from_secs(1));
+ let _ = Pin::new(&mut t1).poll(cx);
}
}
self.state = WorkerState::Shutdown(t1, t2, tx);
- return Ok(Async::NotReady);
+ return Poll::Pending;
}
WorkerState::Available => {
loop {
- match self.rx.poll() {
+ match Pin::new(&mut self.rx).poll_next(cx) {
// handle incoming tcp stream
- Ok(Async::Ready(Some(WorkerCommand(msg)))) => {
- match self.check_readiness(false) {
+ Poll::Ready(Some(WorkerCommand(msg))) => {
+ match self.check_readiness(false, cx) {
Ok(true) => {
let guard = self.conns.get();
let _ = self.services[msg.token.0]
@@ -441,18 +436,17 @@ impl Future for Worker {
);
}
}
- return self.poll();
+ return self.poll(cx);
}
- Ok(Async::NotReady) => {
+ Poll::Pending => {
self.state = WorkerState::Available;
- return Ok(Async::NotReady);
+ return Poll::Pending;
}
- Ok(Async::Ready(None)) | Err(_) => return Ok(Async::Ready(())),
+ Poll::Ready(None) => return Poll::Ready(()),
}
}
}
WorkerState::None => panic!(),
};
}
- */
}
diff --git a/actix-server/tests/test_server.rs b/actix-server/tests/test_server.rs
index 2cce4a7d..91048a2c 100644
--- a/actix-server/tests/test_server.rs
+++ b/actix-server/tests/test_server.rs
@@ -1,10 +1,10 @@
-use std::io::Read;
+use std::io::{self, Read};
use std::sync::mpsc;
use std::{net, thread, time};
use actix_codec::{BytesCodec, Framed};
use actix_server::{Io, Server, ServerConfig};
-use actix_service::{new_service_cfg, service_fn, IntoService, ServiceExt};
+use actix_service::{into_service, service_fn, service_fn_config, IntoService};
use bytes::Bytes;
use futures::{Future, FutureExt, Sink, SinkExt};
use net2::TcpBuilder;
@@ -29,7 +29,7 @@ fn test_bind() {
let sys = actix_rt::System::new("test");
let srv = Server::build()
.bind("test", addr, move || {
- new_service_cfg(move |cfg: &ServerConfig| {
+ service_fn_config(move |cfg: &ServerConfig| {
assert_eq!(cfg.local_addr(), addr);
ok::<_, ()>((|_| ok::<_, ()>(())).into_service())
})
@@ -77,7 +77,7 @@ fn test_listen() {
let lst = net::TcpListener::bind(addr).unwrap();
let srv = Server::build()
.listen("test", lst, move || {
- new_service_cfg(move |cfg: &ServerConfig| {
+ service_fn_config(move |cfg: &ServerConfig| {
assert_eq!(cfg.local_addr(), addr);
ok::<_, ()>((|_| ok::<_, ()>(())).into_service())
})
@@ -95,70 +95,75 @@ fn test_listen() {
let _ = h.join();
}
-#[test]
-#[cfg(unix)]
-fn test_start() {
- let addr = unused_addr();
- let (tx, rx) = mpsc::channel();
+// #[test]
+// #[cfg(unix)]
+// fn test_start() {
+// let addr = unused_addr();
+// let (tx, rx) = mpsc::channel();
- let h = thread::spawn(move || {
- let sys = actix_rt::System::new("test");
- let srv = Server::build()
- .backlog(100)
- .bind("test", addr, move || {
- new_service_cfg(move |cfg: &ServerConfig| {
- assert_eq!(cfg.local_addr(), addr);
+// let h = thread::spawn(move || {
+// let sys = actix_rt::System::new("test");
+// let srv: Server = Server::build()
+// .backlog(100)
+// .bind("test", addr, move || {
+// service_fn_config(move |cfg: &ServerConfig| {
+// assert_eq!(cfg.local_addr(), addr);
- let serv_creator = (move |io: Io<TcpStream>| async {
- panic!("Stream");
- let mut f = Framed::new(io.into_parts().0, BytesCodec);
- f.send(Bytes::from_static(b"test")).await.unwrap();
- Ok::<_, ()>(())
- }).into_service();
+// let srv = into_service(
+// (|io: Io<TcpStream>| {
+// let t = async {
+// let mut f = Framed::new(io.into_parts().0, BytesCodec);
+// f.send(Bytes::from_static(b"test")).await.unwrap();
+// Ok::<_, ()>(())
+// };
+// //ok::<_, ()>(())
+// t
+// }),
+// );
- ok::<_, ()>(serv_creator)
- })
- })
- .unwrap()
- .start();
+// ok::<_, ()>(srv)
+// })
+// })
+// .unwrap()
+// .start();
- let _ = tx.send((srv, actix_rt::System::current()));
- let _ = sys.run();
- });
- let (srv, sys) = rx.recv().unwrap();
+// let _ = tx.send((srv, actix_rt::System::current()));
+// let _ = sys.run();
+// });
+// let (srv, sys) = rx.recv().unwrap();
- let mut buf = [1u8; 4];
- let mut conn = net::TcpStream::connect(addr).unwrap();
- let _ = conn.read_exact(&mut buf);
- assert_eq!(buf, b"test"[..]);
+// let mut buf = [1u8; 4];
+// let mut conn = net::TcpStream::connect(addr).unwrap();
+// let _ = conn.read_exact(&mut buf);
+// assert_eq!(buf, b"test"[..]);
- // pause
- let _ = srv.pause();
- thread::sleep(time::Duration::from_millis(200));
- let mut conn = net::TcpStream::connect(addr).unwrap();
- conn.set_read_timeout(Some(time::Duration::from_millis(100)))
- .unwrap();
- let res = conn.read_exact(&mut buf);
- assert!(res.is_err());
+// // pause
+// let _ = srv.pause();
+// thread::sleep(time::Duration::from_millis(200));
+// let mut conn = net::TcpStream::connect(addr).unwrap();
+// conn.set_read_timeout(Some(time::Duration::from_millis(100)))
+// .unwrap();
+// let res = conn.read_exact(&mut buf);
+// assert!(res.is_err());
- // resume
- let _ = srv.resume();
- thread::sleep(time::Duration::from_millis(100));
- assert!(net::TcpStream::connect(addr).is_ok());
- assert!(net::TcpStream::connect(addr).is_ok());
- assert!(net::TcpStream::connect(addr).is_ok());
+// // resume
+// let _ = srv.resume();
+// thread::sleep(time::Duration::from_millis(100));
+// assert!(net::TcpStream::connect(addr).is_ok());
+// assert!(net::TcpStream::connect(addr).is_ok());
+// assert!(net::TcpStream::connect(addr).is_ok());
- let mut buf = [0u8; 4];
- let mut conn = net::TcpStream::connect(addr).unwrap();
- let _ = conn.read_exact(&mut buf);
- assert_eq!(buf, b"test"[..]);
+// let mut buf = [0u8; 4];
+// let mut conn = net::TcpStream::connect(addr).unwrap();
+// let _ = conn.read_exact(&mut buf);
+// assert_eq!(buf, b"test"[..]);
- // stop
- let _ = srv.stop(false);
- thread::sleep(time::Duration::from_millis(100));
- assert!(net::TcpStream::connect(addr).is_err());
+// // stop
+// let _ = srv.stop(false);
+// thread::sleep(time::Duration::from_millis(100));
+// assert!(net::TcpStream::connect(addr).is_err());
- thread::sleep(time::Duration::from_millis(100));
- let _ = sys.stop();
- let _ = h.join();
-}
+// thread::sleep(time::Duration::from_millis(100));
+// let _ = sys.stop();
+// let _ = h.join();
+// }