Function run_server

redis-server/src/main.rs:184–246

(
    cache_size: usize,
    cache_path: PathBuf,
    durable_fs: bool,
    addr: net::SocketAddr,
    mut shutdown_rx: oneshot::Receiver<()>,
)

}

async fn run_server(
    cache_size: usize,
    cache_path: PathBuf,
    durable_fs: bool,
    addr: net::SocketAddr,
    mut shutdown_rx: oneshot::Receiver<()>,
) {
    info!(%cache_size, ?cache_path, %addr, "Starting with parameters.");

    // Setup the cache here.
    let cache = match ArcDiskCache::new(cache_size, &cache_path, durable_fs) {
        Ok(l) => Arc::new(l),
        Err(err) => {
            error!(?err, "Could not create Arc Disk Cache");
            return;
        }
    };

    let listener = match TcpListener::bind(&addr).await {
        Ok(l) => l,
        Err(e) => {
            error!("Could not bind to redis server address {} -> {:?}", addr, e);
            return;
        }
    };

    let (tx, _rx) = broadcast::channel(1);

    trace!("Listening on {:?}", addr);

    loop {
        tokio::select! {
            Ok(()) = &mut shutdown_rx => {
                tx.send(())
                    .expect("Unable to broadcast shutdown!");
                break;
            }
            res = listener.accept() => {
                match res {
                    Ok((tcpstream, client_socket_addr)) => {
                        tcpstream.set_nodelay(true).expect("Unable to set no delay");
                        // Start the event
                        let (r, w) = tokio::io::split(tcpstream);
                        let r = FramedRead::new(r, RedisCodec::new(cache.clone()));
                        let w = FramedWrite::new(w, RedisCodec::new(cache.clone()));
                        let c_rx = tx.subscribe();
                        let c_cache = cache.clone();
                        // Let it rip.
                        tokio::spawn(client_process(r, w, client_socket_addr, c_rx, c_cache));
                    }
                    Err(e) => {
                        error!("TCP acceptor error, continuing -> {:?}", e);
                    }
                }
            }
        }
    }