feophant 0.9.0

A SQL database server written in Rust and inspired by PostreSQL
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
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135

use crate::{
    codec::{NetworkFrame, PgCodec},
    engine::{
        io::block_layer::file_manager2::{FileManager2, FileManager2Error},
        transactions::TransactionManager,
        Engine,
    },
    processor::ClientProcessor,
};
use futures::{SinkExt, StreamExt};
use std::{ffi::OsString, sync::Arc};
use thiserror::Error;
use tokio::{
    net::TcpListener,
    sync::oneshot::{error::RecvError, Sender},
};
use tokio::{
    pin,
    sync::mpsc::{error::SendError, UnboundedReceiver},
};
use tokio_util::codec::Framed;

pub struct FeOphant {
    pub port: u16,
    listener: TcpListener,
    transaction_manager: TransactionManager,
    engine: Engine,
    file_manager: Arc<FileManager2>,
}

impl FeOphant {
    pub async fn new(data_dir: OsString, port: u16) -> Result<FeOphant, FeOphantError> {
        let file_manager = Arc::new(FileManager2::new(data_dir)?);
        let transaction_manager = TransactionManager::new();
        let engine = Engine::new(file_manager.clone(), transaction_manager.clone());

        let listener = TcpListener::bind(format!("{}{}", "127.0.0.1:", port)).await?;
        let port = listener.local_addr()?.port();
        debug!("Bound to port {0}, but not processing yet.", port);

        Ok(FeOphant {
            port,
            listener,
            transaction_manager,
            engine,
            file_manager,
        })
    }

    /// Starts up the actual server, should be started as its own task
    /// Send on the shutdown_recv to shut it down.
    pub async fn start(&self, shutdown_recv: UnboundedReceiver<Sender<()>>) {
        let shutdown_sender: Option<Sender<()>>;
        info!("Up and listening on port {}", self.port);

        let listen = &self.listener;
        pin!(shutdown_recv);
        pin!(listen);

        loop {
            tokio::select! {
                biased;
                shut_sender = shutdown_recv.recv() => {
                    if let Some(sender) = shut_sender {
                        shutdown_sender = Some(sender);
                        info!("Got shutdown request");
                        break;
                    }
                }
                listen_res = listen.accept() => {
                    if let Ok((stream, client_addr)) = listen_res {
                        info!("Got a connection from {}", client_addr);
                        let tm = self.transaction_manager.clone();
                        let eng = self.engine.clone();
                        tokio::spawn(async move {
                            let codec = PgCodec {};
                            let (mut sink, mut input) = Framed::new(stream, codec).split();

                            let mut process = ClientProcessor::new(eng, tm);
                            while let Some(Ok(event)) = input.next().await {
                                let responses: Vec<NetworkFrame> = match process.process(event).await {
                                    Ok(responses) => responses,
                                    Err(e) => {
                                        warn!("Had a processing error {}", e);
                                        break;
                                    }
                                };

                                for response in responses {
                                    match sink.send(response).await {
                                        Ok(_) => {}
                                        Err(e) => {
                                            warn!("Unable to send response {}", e);
                                            break;
                                        }
                                    }
                                }
                            }
                        });
                    } else if let Err(e) = listen_res {
                        error!("Got error receiving a connection. {0}", e);
                    }
                }
            };
        }

        //Clean up
        match shutdown_sender {
            Some(s) => {
                debug!("Attempting to signal shutdown.");
                s.send(())
                    .unwrap_or_else(|_| warn!("Unable to signal shutdown."));
            }
            None => {
                error!("Exitting before shutting down all the sockets!");
            }
        }
    }
}

#[derive(Debug, Error)]
pub enum FeOphantError {
    #[error("FeOphant already started.")]
    AlreadyStarted(),
    #[error("Can't start the FeOphant twice")]
    CantStartTwice(),
    #[error(transparent)]
    FileManager2Error(#[from] FileManager2Error),
    #[error(transparent)]
    IOError(#[from] std::io::Error),
    #[error(transparent)]
    RecvError(#[from] RecvError),
    #[error(transparent)]
    ShutdownSendError(#[from] SendError<Sender<()>>),
}