rustdis 0.1.1

A partial Redis server implementation intended purely for educational purposes
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

use std::net::SocketAddr;
use std::sync::{Arc, Mutex};
use tokio::io::AsyncWriteExt;
use tokio::net::{TcpListener, TcpStream};
use tracing::{error, info, instrument};

use crate::commands::executable::Executable;
use crate::commands::Command;
use crate::connection::Connection;
use crate::store::Store;
use crate::Error;

const PORT: u16 = 6379;

pub async fn run() -> Result<(), Error> {
    let subscriber = tracing_subscriber::FmtSubscriber::new();
    tracing::subscriber::set_global_default(subscriber)?;

    let listener = TcpListener::bind(("127.0.0.1", PORT)).await?;
    let store = Arc::new(Mutex::new(Store::new()));

    info!("Redis server listening on {}", listener.local_addr()?);

    loop {
        let (socket, client_address) = listener.accept().await?;
        let store = store.clone();
        info!("Accepted connection from {:?}", client_address);

        tokio::spawn(async move {
            if let Err(e) = handle_connection(socket, client_address, store).await {
                error!(e);
            }
        });
    }
}

#[instrument(
    name = "connection",
    skip(stream, store),
    fields(connection_id, client_address)
)]
async fn handle_connection(
    stream: TcpStream,
    client_address: SocketAddr,
    store: Arc<Mutex<Store>>,
) -> Result<(), Error> {
    let mut conn = Connection::new(stream, client_address);

    tracing::Span::current()
        .record("connection_id", conn.id.to_string())
        .record("client_address", client_address.to_string());

    while let Some(frame) = conn.read_frame().await? {
        info!("Received frame from client: {:?}", frame);
        let cmd = Command::try_from(frame)?;
        let res = cmd.exec(store.clone())?;
        info!("Sending response to client: {:?}", res);
        let res: Vec<u8> = res.into();

        conn.writer.write_all(&res).await?;
    }

    info!("Connection closed");
    Ok(())
}