netpong/src/server/mod.rs

#![cfg(feature = "server")]
use std::time::Duration;

use libpt::log::{debug, info, trace, warn};
use tokio::{
    io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
    net::{TcpListener, TcpStream},
    time::timeout,
};

use crate::common::conf::Config;

pub mod errors;
use errors::*;

pub struct Server {
    cfg: Config,
    pub timeout: Option<Duration>,
    server: TcpListener,
}

impl Server {
    pub async fn build(cfg: Config) -> anyhow::Result<Self> {
        let server = TcpListener::bind(cfg.addr).await?;
        let timeout = Some(Duration::from_secs(5));
        Ok(Server {
            cfg,
            timeout,
            server,
        })
    }
    pub async fn run(self) -> anyhow::Result<()> {
        loop {
            let (stream, addr) = match self.server.accept().await {
                Ok(s) => s,
                Err(err) => {
                    warn!("could not accept stream: {err:?}");
                    continue;
                }
            };
            match self.handle_stream(stream).await {
                Ok(_) => (),
                Err(err) => {
                    match err {
                        ServerError::Timeout(_) => {
                            info!("stream {:?} timed out", addr)
                        }
                        _ => {
                            warn!("error while handling stream: {:?}", err)
                        }
                    }
                    continue;
                }
            };
        }
    }

    async fn handle_stream(&self, stream: TcpStream) -> Result<()> {
        let mut pings: usize = 0;
        let addr = match stream.peer_addr() {
            Ok(a) => a,
            Err(err) => {
                debug!("could not get peer address: {:?}", err);
                return Err(err.into());
            }
        };
        info!("new peer: {:?}", addr);
        let mut buf = Vec::new();
        let mut reader = BufReader::new(stream);
        let mut len;
        loop {
            len = match self.read(&mut reader, &mut buf).await {
                Ok(len) => len,
                Err(err) => {
                    match err {
                        ServerError::Timeout(_) => {
                            info!("peer {:?} timed out", addr)
                        }
                        _ => return Err(err),
                    }
                    break;
                }
            };
            trace!("received message: {:X?}", buf);
            if len == 0 {
                trace!("len is apperently 0: {len:?}");
                break;
            } else {
                let msg = self.decode(&buf)?;
                info!("< {:?} : {}", addr, msg);
                if msg.contains("ping") {
                    pings += 1;
                }
                if pings < self.cfg.win_after {
                    reader.write_all(b"pong\0").await?;
                    info!("> {:?} : pong", addr,);
                } else {
                    reader.write_all(b"you win!\0").await?;
                    info!("> {:?} : you win!", addr,);
                    reader.shutdown().await?;
                    break;
                }
                buf.clear();

                // we should wait, so that we don't spam the client
                std::thread::sleep(self.cfg.delay);
            }
        }
        info!("disconnected peer: {:?}", addr);
        Ok(())
    }

    #[inline]
    fn decode(&self, buf: &Vec<u8>) -> Result<String> {
        Ok(String::from_utf8(buf.clone())?.replace('\n', "\\n"))
    }

    #[inline]
    async fn read(&self, reader: &mut BufReader<TcpStream>, buf: &mut Vec<u8>) -> Result<usize> {
        match timeout(self.cfg.timeout, reader.read_until(0x00, buf)).await? {
            Ok(len) => Ok(len),
            Err(err) => Err(err.into()),
        }
    }
}