Use a builder instead of a raw function
This pattern will, in the future, allow the registration of events handlers and context objects.
This commit is contained in:
parent
c25dfb253a
commit
98e4beadd3
3 changed files with 75 additions and 70 deletions
|
@ -1,6 +1,7 @@
|
||||||
use env_logger::{Builder, Env};
|
use env_logger::{Builder, Env};
|
||||||
|
use opensmtpd::SmtpIn;
|
||||||
|
|
||||||
fn main() {
|
fn main() {
|
||||||
Builder::from_env(Env::default().default_filter_or("debug")).init();
|
Builder::from_env(Env::default().default_filter_or("debug")).init();
|
||||||
opensmtpd::run();
|
SmtpIn::new().run();
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,8 +1,6 @@
|
||||||
use crate::errors::Error;
|
use crate::errors::Error;
|
||||||
use nom::{
|
use nom::{alt, alt_complete, call, complete, cond, do_parse, error_position, map_res, named, tag,
|
||||||
alt, alt_complete, call, complete, cond, do_parse, error_position, map_res, named, tag,
|
take_until, take_while};
|
||||||
take_until, take_while,
|
|
||||||
};
|
|
||||||
|
|
||||||
#[derive(Debug, PartialEq)]
|
#[derive(Debug, PartialEq)]
|
||||||
pub enum Kind {
|
pub enum Kind {
|
||||||
|
|
64
src/lib.rs
64
src/lib.rs
|
@ -9,25 +9,27 @@ use std::io;
|
||||||
use std::sync::mpsc;
|
use std::sync::mpsc;
|
||||||
use std::thread;
|
use std::thread;
|
||||||
|
|
||||||
/// Read a line from the standard input.
|
pub struct SmtpIn {
|
||||||
/// Since EOF should not append, it is considered as an error.
|
sessions: HashMap<u64, (mpsc::Sender<Entry>, thread::JoinHandle<()>)>,
|
||||||
fn read() -> Result<String, Error> {
|
}
|
||||||
|
|
||||||
|
impl SmtpIn {
|
||||||
|
/// Read a line from the standard input.
|
||||||
|
/// Since EOF should not append, it is considered as an error.
|
||||||
|
fn read() -> Result<String, Error> {
|
||||||
let mut input = String::new();
|
let mut input = String::new();
|
||||||
let nb = io::stdin().read_line(&mut input)?;
|
let nb = io::stdin().read_line(&mut input)?;
|
||||||
match nb {
|
match nb {
|
||||||
0 => Err(Error::new("end of file")),
|
0 => Err(Error::new("end of file")),
|
||||||
_ => Ok(input),
|
_ => Ok(input),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Dispatch the entry into its session's thread. If such thread does not
|
/// Dispatch the entry into its session's thread. If such thread does not
|
||||||
/// already exists, creates it.
|
/// already exists, creates it.
|
||||||
fn dispatch(
|
fn dispatch(&mut self, input: &str) -> Result<(), Error> {
|
||||||
sessions: &mut HashMap<u64, (mpsc::Sender<Entry>, thread::JoinHandle<()>)>,
|
|
||||||
input: &str,
|
|
||||||
) -> Result<(), Error> {
|
|
||||||
let entry = Entry::from_str(input)?;
|
let entry = Entry::from_str(input)?;
|
||||||
let channel = match sessions.get(&entry.session_id) {
|
let channel = match self.sessions.get(&entry.session_id) {
|
||||||
Some((r, _)) => r,
|
Some((r, _)) => r,
|
||||||
None => {
|
None => {
|
||||||
let (tx, rx) = mpsc::channel();
|
let (tx, rx) = mpsc::channel();
|
||||||
|
@ -41,46 +43,50 @@ fn dispatch(
|
||||||
debug!("thread {}: {:?}", thread::current().name().unwrap(), e);
|
debug!("thread {}: {:?}", thread::current().name().unwrap(), e);
|
||||||
}
|
}
|
||||||
})?;
|
})?;
|
||||||
sessions.insert(entry.session_id, (tx, handle));
|
self.sessions.insert(entry.session_id, (tx, handle));
|
||||||
let (r, _) = sessions.get(&entry.session_id).unwrap();
|
let (r, _) = self.sessions.get(&entry.session_id).unwrap();
|
||||||
r
|
r
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
channel.send(entry)?;
|
channel.send(entry)?;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Allow each child thread to exit gracefully. First, the session table is
|
/// Allow each child thread to exit gracefully. First, the session table is
|
||||||
/// drained so all the references to the senders are dropped, which will
|
/// drained so all the references to the senders are dropped, which will
|
||||||
/// cause the receivers threads to exit. Then, we uses the join handlers in
|
/// cause the receivers threads to exit. Then, we uses the join handlers in
|
||||||
/// order to wait for the actual exit.
|
/// order to wait for the actual exit.
|
||||||
fn graceful_exit_children(
|
fn graceful_exit_children(&mut self) {
|
||||||
sessions: &mut HashMap<u64, (mpsc::Sender<Entry>, thread::JoinHandle<()>)>,
|
|
||||||
) {
|
|
||||||
let mut handles = Vec::new();
|
let mut handles = Vec::new();
|
||||||
for (_, (_, h)) in sessions.drain() {
|
for (_, (_, h)) in self.sessions.drain() {
|
||||||
handles.push(h);
|
handles.push(h);
|
||||||
}
|
}
|
||||||
for h in handles {
|
for h in handles {
|
||||||
let _ = h.join();
|
let _ = h.join();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Run the infinite loop that will read and process input from stdin.
|
pub fn new() -> Self {
|
||||||
pub fn run() {
|
SmtpIn {
|
||||||
let mut sessions = HashMap::new();
|
sessions: HashMap::new(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Run the infinite loop that will read and process input from stdin.
|
||||||
|
pub fn run(&mut self) {
|
||||||
loop {
|
loop {
|
||||||
let line = match read() {
|
let line = match SmtpIn::read() {
|
||||||
Ok(l) => l,
|
Ok(l) => l,
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
graceful_exit_children(&mut sessions);
|
self.graceful_exit_children();
|
||||||
error!("{}", e);
|
error!("{}", e);
|
||||||
std::process::exit(1);
|
std::process::exit(1);
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
match dispatch(&mut sessions, &line) {
|
match self.dispatch(&line) {
|
||||||
Ok(_) => {}
|
Ok(_) => {}
|
||||||
Err(e) => warn!("{}", e),
|
Err(e) => warn!("{}", e),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
Reference in a new issue