Active revoker

Cargo.toml

[package]
name = "active-revok"
version = "0.1.0"
edition = "2021"

[dependencies]
tokio = { version = "1.42.0", features = ["rt-multi-thread", "macros", "test-util", "sync"] }
log = "0.4.22"
env_logger = "0.11.5"

lib.rs

use std::sync::Arc;
use tokio::sync::RwLock;

struct Revoker {
    rx: RwLock<tokio::sync::mpsc::Receiver<()>>,
    tx: tokio::sync::mpsc::Sender<()>,
    revoked: RwLock<bool>,
}

impl Revoker {
    fn new(rx: tokio::sync::mpsc::Receiver<()>, tx: tokio::sync::mpsc::Sender<()>) -> Self {
        Self {
            rx: RwLock::new(rx),
            tx,
            revoked: RwLock::new(false),
        }
    }
}

impl Revoker {
    async fn revoked(&self) -> bool {
        *self.revoked.read().await
    }

    async fn run(&self) {
        if self.rx.write().await.recv().await.is_some() {
            log::info!("Revocation des accès");
            *self.revoked.write().await = true;
            self.tx.send(()).await.unwrap();
        }
    }
}

struct Server {
    rx: RwLock<tokio::sync::mpsc::Receiver<String>>,
    tx: tokio::sync::mpsc::Sender<String>,
    revoker: Arc<Revoker>,
}

impl Server {
    fn new(
        rx: tokio::sync::mpsc::Receiver<String>,
        tx: tokio::sync::mpsc::Sender<String>,
        revoker: Arc<Revoker>,
    ) -> Self {
        Self {
            rx: RwLock::new(rx),
            tx,
            revoker,
        }
    }
}

impl Server {
    async fn run(&self) {
        while let Some(msg) = self.rx.write().await.recv().await {
            let result = match msg.as_str() {
                "stop" => break,
                _ => match self.revoker.revoked().await {
                    true => {
                        log::warn!("Accès révoqué");
                        "Ko"
                    }
                    false => {
                        log::info!("Accès autorisé");
                        "Ok"
                    }
                },
            };
            self.tx
                .send(result.to_string())
                .await
                .expect("Failed to send message");
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tokio::join;
    use tokio::sync::mpsc::{Receiver, Sender};

    #[tokio::test]
    async fn it_works() {
        let _ = env_logger::try_init();

        let (tx_revoker, rx_revoker) = tokio::sync::mpsc::channel(1);
        let (tx_revoker_response, mut rx_revoker_response) = tokio::sync::mpsc::channel(1);
        let revoker = Arc::new(Revoker::new(rx_revoker, tx_revoker_response));
        let (tx_server, rx_server) = tokio::sync::mpsc::channel(1);
        let (tx_client, mut rx_client) = tokio::sync::mpsc::channel(1);
        let server = Server::new(rx_server, tx_client, revoker.clone());

        async fn send_recv(
            message: &str,
            tx: &Sender<String>,
            rx: &mut Receiver<String>,
            status: bool,
        ) {
            tx.send(message.to_string()).await.unwrap();
            let response = rx.recv().await.unwrap();
            if status {
                assert_eq!(response, "Ok");
            } else {
                assert_eq!(response, "Ko");
            }
        }

        let tests = async move {
            send_recv("req", &tx_server, &mut rx_client, true).await;
            send_recv("req", &tx_server, &mut rx_client, true).await;
            send_recv("req", &tx_server, &mut rx_client, true).await;
            // revoke
            tx_revoker.send(()).await.unwrap();
            assert!(rx_revoker_response.recv().await.is_some());
            send_recv("req", &tx_server, &mut rx_client, false).await;
            send_recv("req", &tx_server, &mut rx_client, false).await;
            // stop
            tx_server.send("stop".to_string()).await.unwrap();
        };

        join!(server.run(), revoker.run(), tests);
    }
}