rxqlite 0.1.15

A secured distributed sqlite database built upon `openraft`, `sqlx` and `sqlite`.
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
136
137
138
139
140
141
142
143
144
145
146
147
148

use std::sync::Arc;

use warp::reply;

use crate::app::App;
use crate::Node;
use crate::TypeConfig;
use openraft::LeaderId;
use openraft::LogId;
use rxqlite_common::{Message,MessageResponse};
use crate::convert::RXQliteInto;

pub async fn sql_consistent_or_fast(
    message: Message,
    app: Arc<App>,
    consistent: bool,
) -> Result::<
            openraft::raft::ClientWriteResponse<TypeConfig>,
            openraft::error::RaftError<u64, openraft::error::ClientWriteError<u64, Node>>,
        > {
    let sql = message.sql();

    let is_write = rxqlite_sqlx_common::is_query_write(sql);
    if let Err(err) = &is_write {
      let response_message = MessageResponse::Error(format!("{}",err));
      let client_write_response = openraft::raft::ClientWriteResponse::<TypeConfig> {
            log_id: LogId {
                leader_id: LeaderId {
                    term: u64::MAX,
                    node_id: u64::MAX,
                },
                index: u64::MAX,
            },
            data: Some(response_message),
            membership: None,
        };
        let res = Result::<
            openraft::raft::ClientWriteResponse<TypeConfig>,
            openraft::error::RaftError<u64, openraft::error::ClientWriteError<u64, Node>>,
        >::Ok(client_write_response);
        return res;
    }
    let is_write=is_write.unwrap();
    if is_write {
        let res: Result<openraft::raft::ClientWriteResponse<TypeConfig>, _> =
            app.raft.client_write(message).await;
        res
    } else {
        let do_it_locally = if consistent {
            if let Ok(_read_log_id) = app.raft.ensure_linearizable().await {
                true
            } else {
                false
            }
        } else {
            true
        };
        if do_it_locally {
            let sqlite_and_path = app.sqlite_and_path.read().await;
            let response_message = rxqlite_sqlx_common::do_sql(&sqlite_and_path, message).await;
            let client_write_response = openraft::raft::ClientWriteResponse::<TypeConfig> {
                log_id: LogId {
                    leader_id: LeaderId {
                        term: u64::MAX,
                        node_id: u64::MAX,
                    },
                    index: u64::MAX,
                },
                data: Some(response_message),
                membership: None,
            };
            let res = Result::<
                openraft::raft::ClientWriteResponse<TypeConfig>,
                openraft::error::RaftError<u64, openraft::error::ClientWriteError<u64, Node>>,
            >::Ok(client_write_response);
            res
        } else {
            let server_metrics = app.raft.server_metrics().borrow().clone();
            if let Some(leader_id) = server_metrics.current_leader {
                let membership_config = server_metrics.membership_config;
                let leader_node = membership_config.nodes().find_map(|(node_id, node)| {
                    if node_id == &leader_id {
                        Some(node.clone())
                    } else {
                        None
                    }
                });
                let res = Result::<
                    openraft::raft::ClientWriteResponse<TypeConfig>,
                    openraft::error::RaftError<u64, openraft::error::ClientWriteError<u64, Node>>,
                >::Err(openraft::error::RaftError::APIError(
                    openraft::error::ClientWriteError::ForwardToLeader(
                        openraft::error::ForwardToLeader {
                            leader_id: Some(leader_id),
                            leader_node: leader_node,
                        },
                    ),
                ));
                res
            } else {
                let res = Result::<
                    openraft::raft::ClientWriteResponse<TypeConfig>,
                    openraft::error::RaftError<u64, openraft::error::ClientWriteError<u64, Node>>,
                >::Err(openraft::error::RaftError::APIError(
                    openraft::error::ClientWriteError::ForwardToLeader(
                        openraft::error::ForwardToLeader {
                            leader_id: None,
                            leader_node: None,
                        },
                    ),
                ));
                res
            }
        }
    }
}

pub async fn sql(
    message: Message,
    app: Arc<App>,
) -> Result<impl warp::Reply, std::convert::Infallible> {
    let res=sql_consistent_or_fast(message, app, false).await;
    Ok(reply::json(&res))
}

pub async fn sql_consistent(
    message: Message,
    app: Arc<App>,
) -> Result<impl warp::Reply, std::convert::Infallible> {
    let res=sql_consistent_or_fast(message, app, true).await;
    Ok(reply::json(&res))
}

pub async fn lite_sql(
    message: Message,
    app: Arc<App>,
) -> Result<impl warp::Reply, std::convert::Infallible> {
    let res=sql_consistent_or_fast(message, app, false).await.into_lite();
    Ok(reply::json(&res))
}

pub async fn lite_sql_consistent(
    message: Message,
    app: Arc<App>,
) -> Result<impl warp::Reply, std::convert::Infallible> {
    let res=sql_consistent_or_fast(message, app, true).await.into_lite();
    Ok(reply::json(&res))
}