Skip to content

Commit 42ec345

Browse files
fafhrd91fafhrd91
committed
add signals support
1 parent e6daca7 commit 42ec345

File tree

4 files changed

+254
-120
lines changed

4 files changed

+254
-120
lines changed

‎actix-server/src/builder.rs‎

Lines changed: 128 additions & 117 deletions
Original file line numberDiff line numberDiff line change
@@ -11,15 +11,14 @@ use net2::TcpBuilder;
1111
use num_cpus;
1212
use tokio_timer::sleep;
1313

14-
// use actix::{actors::signal};
15-
16-
use super::accept::{AcceptLoop, AcceptNotify, Command};
17-
use super::config::{ConfiguredService, ServiceConfig};
18-
use super::server::{Server, ServerCommand};
19-
use super::services::{InternalServiceFactory, StreamNewService, StreamServiceFactory};
20-
use super::services::{ServiceFactory, ServiceNewService};
21-
use super::worker::{self, Worker, WorkerAvailability, WorkerClient};
22-
use super::Token;
14+
use crate::accept::{AcceptLoop, AcceptNotify, Command};
15+
use crate::config::{ConfiguredService, ServiceConfig};
16+
use crate::server::{Server, ServerCommand};
17+
use crate::services::{InternalServiceFactory, StreamNewService, StreamServiceFactory};
18+
use crate::services::{ServiceFactory, ServiceNewService};
19+
use crate::signals::{Signal, Signals};
20+
use crate::worker::{self, Worker, WorkerAvailability, WorkerClient};
21+
use crate::Token;
2322

2423
/// Server builder
2524
pub struct ServerBuilder {
@@ -244,11 +243,12 @@ impl ServerBuilder {
244243
self.accept
245244
.start(mem::replace(&mut self.sockets, Vec::new()), workers);
246245

246+
// handle signals
247+
if !self.no_signals {
248+
Signals::start(self.server.clone());
249+
}
250+
247251
// start http server actor
248-
// let signals = self.subscribe_to_signals();
249-
// if let Some(signals) = signals {
250-
// signals.do_send(signal::Subscribe(addr.clone().recipient()))
251-
// }
252252
let server = self.server.clone();
253253
spawn(self);
254254
server
@@ -271,124 +271,135 @@ impl ServerBuilder {
271271

272272
worker
273273
}
274-
}
275-
276-
// /// Signals support
277-
// /// Handle `SIGINT`, `SIGTERM`, `SIGQUIT` signals and stop actix system
278-
// /// message to `System` actor.
279-
// impl Handler<signal::Signal> for Server {
280-
// type Result = ();
281-
282-
// fn handle(&mut self, msg: signal::Signal, ctx: &mut Context<Self>) {
283-
// match msg.0 {
284-
// signal::SignalType::Int => {
285-
// info!("SIGINT received, exiting");
286-
// self.exit = true;
287-
// Handler::<StopServer>::handle(self, StopServer { graceful: false }, ctx);
288-
// }
289-
// signal::SignalType::Term => {
290-
// info!("SIGTERM received, stopping");
291-
// self.exit = true;
292-
// Handler::<StopServer>::handle(self, StopServer { graceful: true }, ctx);
293-
// }
294-
// signal::SignalType::Quit => {
295-
// info!("SIGQUIT received, exiting");
296-
// self.exit = true;
297-
// Handler::<StopServer>::handle(self, StopServer { graceful: false }, ctx);
298-
// }
299-
// _ => (),
300-
// }
301-
// }
302-
// }
303-
304-
impl Future for ServerBuilder {
305-
type Item = ();
306-
type Error = ();
307274

308-
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
309-
loop {
310-
match self.cmd.poll() {
311-
Ok(Async::Ready(None)) | Err(_) => return Ok(Async::Ready(())),
312-
Ok(Async::NotReady) => return Ok(Async::NotReady),
313-
Ok(Async::Ready(Some(item))) => match item {
314-
ServerCommand::Pause(tx) => {
315-
self.accept.send(Command::Pause);
316-
let _ = tx.send(());
275+
fn handle_cmd(&mut self, item: ServerCommand) {
276+
match item {
277+
ServerCommand::Pause(tx) => {
278+
self.accept.send(Command::Pause);
279+
let _ = tx.send(());
280+
}
281+
ServerCommand::Resume(tx) => {
282+
self.accept.send(Command::Resume);
283+
let _ = tx.send(());
284+
}
285+
ServerCommand::Signal(sig) => {
286+
// Signals support
287+
// Handle `SIGINT`, `SIGTERM`, `SIGQUIT` signals and stop actix system
288+
match sig {
289+
Signal::Int => {
290+
info!("SIGINT received, exiting");
291+
self.exit = true;
292+
self.handle_cmd(ServerCommand::Stop {
293+
graceful: false,
294+
completion: None,
295+
})
317296
}
318-
ServerCommand::Resume(tx) => {
319-
self.accept.send(Command::Resume);
320-
let _ = tx.send(());
297+
Signal::Term => {
298+
info!("SIGTERM received, stopping");
299+
self.exit = true;
300+
self.handle_cmd(ServerCommand::Stop {
301+
graceful: true,
302+
completion: None,
303+
})
321304
}
322-
ServerCommand::Stop {
323-
graceful,
324-
completion,
325-
} => {
326-
let exit = self.exit;
327-
328-
// stop accept thread
329-
self.accept.send(Command::Stop);
330-
331-
// stop workers
332-
if !self.workers.is_empty() {
333-
spawn(
334-
futures_unordered(
335-
self.workers
336-
.iter()
337-
.map(move |worker| worker.1.stop(graceful)),
338-
)
339-
.collect()
340-
.then(move |_| {
341-
let _ = completion.send(());
342-
if exit {
343-
spawn(sleep(Duration::from_millis(300)).then(|_| {
344-
System::current().stop();
345-
ok(())
346-
}));
347-
}
348-
ok(())
349-
}),
350-
)
351-
} else {
352-
// we need to stop system if server was spawned
353-
if self.exit {
305+
Signal::Quit => {
306+
info!("SIGQUIT received, exiting");
307+
self.exit = true;
308+
self.handle_cmd(ServerCommand::Stop {
309+
graceful: false,
310+
completion: None,
311+
})
312+
}
313+
_ => (),
314+
}
315+
}
316+
ServerCommand::Stop {
317+
graceful,
318+
completion,
319+
} => {
320+
let exit = self.exit;
321+
322+
// stop accept thread
323+
self.accept.send(Command::Stop);
324+
325+
// stop workers
326+
if !self.workers.is_empty() {
327+
spawn(
328+
futures_unordered(
329+
self.workers
330+
.iter()
331+
.map(move |worker| worker.1.stop(graceful)),
332+
)
333+
.collect()
334+
.then(move |_| {
335+
if let Some(tx) = completion {
336+
let _ = tx.send(());
337+
}
338+
if exit {
354339
spawn(sleep(Duration::from_millis(300)).then(|_| {
355340
System::current().stop();
356341
ok(())
357342
}));
358343
}
359-
let _ = completion.send(());
360-
}
344+
ok(())
345+
}),
346+
)
347+
} else {
348+
// we need to stop system if server was spawned
349+
if self.exit {
350+
spawn(sleep(Duration::from_millis(300)).then(|_| {
351+
System::current().stop();
352+
ok(())
353+
}));
361354
}
362-
ServerCommand::WorkerDied(idx) => {
363-
let mut found = false;
355+
if let Some(tx) = completion {
356+
let _ = tx.send(());
357+
}
358+
}
359+
}
360+
ServerCommand::WorkerDied(idx) => {
361+
let mut found = false;
362+
for i in 0..self.workers.len() {
363+
if self.workers[i].0 == idx {
364+
self.workers.swap_remove(i);
365+
found = true;
366+
break;
367+
}
368+
}
369+
370+
if found {
371+
error!("Worker has died {:?}, restarting", idx);
372+
373+
let mut new_idx = self.workers.len();
374+
'found: loop {
364375
for i in 0..self.workers.len() {
365-
if self.workers[i].0 == idx {
366-
self.workers.swap_remove(i);
367-
found = true;
368-
break;
376+
if self.workers[i].0 == new_idx {
377+
new_idx += 1;
378+
continue 'found;
369379
}
370380
}
381+
break;
382+
}
371383

372-
if found {
373-
error!("Worker has died {:?}, restarting", idx);
374-
375-
let mut new_idx = self.workers.len();
376-
'found: loop {
377-
for i in 0..self.workers.len() {
378-
if self.workers[i].0 == new_idx {
379-
new_idx += 1;
380-
continue 'found;
381-
}
382-
}
383-
break;
384-
}
384+
let worker = self.start_worker(new_idx, self.accept.get_notify());
385+
self.workers.push((new_idx, worker.clone()));
386+
self.accept.send(Command::Worker(worker));
387+
}
388+
}
389+
}
390+
}
391+
}
385392

386-
let worker = self.start_worker(new_idx, self.accept.get_notify());
387-
self.workers.push((new_idx, worker.clone()));
388-
self.accept.send(Command::Worker(worker));
389-
}
390-
}
391-
},
393+
impl Future for ServerBuilder {
394+
type Item = ();
395+
type Error = ();
396+
397+
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
398+
loop {
399+
match self.cmd.poll() {
400+
Ok(Async::Ready(None)) | Err(_) => return Ok(Async::Ready(())),
401+
Ok(Async::NotReady) => return Ok(Async::NotReady),
402+
Ok(Async::Ready(Some(item))) => self.handle_cmd(item),
392403
}
393404
}
394405
}

‎actix-server/src/lib.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ mod config;
66
mod counter;
77
mod server;
88
mod services;
9+
mod signals;
910
pub mod ssl;
1011
mod worker;
1112

‎actix-server/src/server.rs‎

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,16 +2,18 @@ use futures::sync::mpsc::UnboundedSender;
22
use futures::sync::oneshot;
33
use futures::Future;
44

5-
use super::builder::ServerBuilder;
5+
use crate::builder::ServerBuilder;
6+
use crate::signals::Signal;
67

78
pub(crate) enum ServerCommand {
89
WorkerDied(usize),
910
Pause(oneshot::Sender<()>),
1011
Resume(oneshot::Sender<()>),
12+
Signal(Signal),
1113
/// Whether to try and shut down gracefully
1214
Stop {
1315
graceful: bool,
14-
completion: oneshot::Sender<()>,
16+
completion: Option<oneshot::Sender<()>>,
1517
},
1618
}
1719

@@ -28,6 +30,10 @@ impl Server {
2830
ServerBuilder::default()
2931
}
3032

33+
pub(crate) fn signal(&self, sig: Signal) {
34+
let _ = self.0.unbounded_send(ServerCommand::Signal(sig));
35+
}
36+
3137
pub(crate) fn worker_died(&self, idx: usize) {
3238
let _ = self.0.unbounded_send(ServerCommand::WorkerDied(idx));
3339
}
@@ -56,7 +62,7 @@ impl Server {
5662
let (tx, rx) = oneshot::channel();
5763
let _ = self.0.unbounded_send(ServerCommand::Stop {
5864
graceful,
59-
completion: tx,
65+
completion: Some(tx),
6066
});
6167
rx.map_err(|_| ())
6268
}

0 commit comments

Comments
 (0)