Add mail processing scaffolding

After receiving the email we don't want to process it in the same thread
as soon as we can, instead we queue it to a MailProcessor. This allows
us to be more flexible with the pipeline and reduce the amount of
concurrency for the database connection. This also helps with writing
possible fences around resource consumption.

Signed-off-by: Mroik <mroik@delayed.space>

1 files changed, 15 insertions, 4 deletions

diff --git a/src/smtp_server.rs b/src/smtp_server.rs
index 2795ecf..bacf8eb 100644
--- a/src/smtp_server.rs
+++ b/src/smtp_server.rs
@@ -4,7 +4,9 @@ use std::{
 };
 
 use anyhow::Result;
-use tokio::spawn;
+use tokio::{spawn, sync::mpsc::Sender};
+
+use crate::model::Mail;
 
 const SERVER_NAME: &str = "";
 
@@ -22,12 +24,13 @@ impl SmtpServer {
     }
 
     // TODO: trap SIGINT to stop server?
-    pub async fn run(&mut self) -> Result<()> {
+    pub async fn run(&mut self, tx_processor: Sender<Mail>) -> Result<()> {
         self.running = true;
         while self.running {
             let (stream, addr) = self.listener.accept()?;
             let session = SessionHandler {
                 addr,
+                tx_processor: tx_processor.clone(),
                 state: SessionState::default(),
                 client_host: String::new(),
                 from: None,
@@ -42,6 +45,7 @@ impl SmtpServer {
 
 struct SessionHandler {
     addr: SocketAddr,
+    tx_processor: Sender<Mail>,
     state: SessionState,
     client_host: String,
     from: Option<String>,
@@ -158,18 +162,25 @@ impl SessionHandler {
         Ok(Reply::StartMailInput)
     }
 
-    // TODO: Forward and store email
     async fn process_mail(&mut self) -> Result<Reply> {
         if self.from.is_none() || self.to.is_empty() || self.data.trim().is_empty() {
             return Ok(Reply::InvalidCommand);
         }
 
+        let mail = Mail {
+            from: self.from.as_ref().unwrap().clone(),
+            recipient: self.to.clone(),
+            data: self.data.clone(),
+        };
+
+        self.tx_processor.send(mail).await?;
+
         self.state = SessionState::Normal;
         self.from = None;
         self.to.clear();
         self.data.clear();
 
-        todo!()
+        Ok(Reply::Completed(String::from("Ok")))
     }
 }