rnacos 0.8.3

Nacos server re-implemented in Rust.
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

use std::sync::Arc;

use async_raft_ext::raft::ClientWriteRequest;

use self::model::{RouterRequest, RouterResponse};
use super::{db::table::TableManagerAsyncReq, join_node, store::ClientRequest};
use crate::grpc::handler::RAFT_ROUTE_REQUEST;
use crate::grpc::PayloadUtils;
use crate::namespace::model::NamespaceRaftResult;
use crate::raft::network::factory::RaftClusterRequestSender;
use crate::raft::store::ClientResponse;
use crate::transfer::model::TransferImportRequest;
use crate::{
    common::appdata::AppShareData,
    config::core::{ConfigAsyncCmd, ConfigKey},
};

pub mod model;
pub mod route;
pub mod routeapi;

pub async fn handle_route(
    app: &Arc<AppShareData>,
    req: RouterRequest,
) -> anyhow::Result<RouterResponse> {
    match req {
        RouterRequest::RaftRequest(req) => {
            let r = app.raft.client_write(ClientWriteRequest::new(req)).await?;
            Ok(RouterResponse::RaftResponse(r.data))
        }
        RouterRequest::ConfigSet {
            key,
            value,
            op_user,
            config_type,
            desc,
            extend_info: _,
        } => {
            let config_key: ConfigKey = (&key as &str).into();
            app.config_addr
                .send(ConfigAsyncCmd::Add {
                    key: config_key,
                    value,
                    op_user,
                    config_type,
                    desc,
                })
                .await??;
            Ok(RouterResponse::None)
        }
        RouterRequest::ConfigDel {
            key,
            extend_info: _,
        } => {
            let config_key: ConfigKey = (&key as &str).into();
            app.config_addr
                .send(ConfigAsyncCmd::Delete(config_key))
                .await??;
            Ok(RouterResponse::None)
        }
        RouterRequest::JoinNode {
            node_id,
            node_addr: addr,
        } => {
            app.raft
                .client_write(ClientWriteRequest::new(ClientRequest::NodeAddr {
                    id: node_id,
                    addr,
                }))
                .await?;
            app.raft.add_non_voter(node_id).await?;
            join_node(app.raft.as_ref(), app.raft_store.as_ref(), node_id).await?;
            Ok(RouterResponse::None)
        }
        RouterRequest::TableManagerReq { req } => {
            let result = app
                .raft_table_manage
                .send(TableManagerAsyncReq(req))
                .await??;
            Ok(RouterResponse::TableManagerResult { result })
        }
        RouterRequest::TableManagerQueryReq { req } => {
            let result = app.raft_table_manage.send(req).await??;
            Ok(RouterResponse::TableManagerResult { result })
        }
        RouterRequest::CacheLimiterReq { req } => {
            let result = app.cache_manager.send(req).await??;
            Ok(RouterResponse::CacheManagerResult { result })
        }
        RouterRequest::NamespaceReq { req } => {
            let resp = app
                .raft
                .client_write(ClientWriteRequest::new(ClientRequest::NamespaceReq(req)))
                .await?;
            if let ClientResponse::Success = resp.data {
                return Ok(RouterResponse::NamespaceResult {
                    result: NamespaceRaftResult::None,
                });
            }
            Ok(RouterResponse::None)
        }
        RouterRequest::ImportData { data, param } => {
            let result = app
                .transfer_import_manager
                .send(TransferImportRequest::Import(data, param))
                .await??;
            Ok(RouterResponse::ImportResult { result })
        }
        RouterRequest::CacheQuery { req } => {
            let result = app.direct_cache_manager.send(req).await??;
            Ok(RouterResponse::CacheQueryResult { result })
        }
    }
}

pub async fn router_request(
    req: RouterRequest,
    addr: Arc<String>,
    cluster_sender: &Arc<RaftClusterRequestSender>,
) -> anyhow::Result<RouterResponse> {
    let request = serde_json::to_string(&req).unwrap_or_default();
    let payload = PayloadUtils::build_payload(RAFT_ROUTE_REQUEST, request);
    let resp_payload = cluster_sender.send_request(addr, payload).await?;
    let body_vec = resp_payload.body.unwrap_or_default().value;
    let router_resp: RouterResponse = serde_json::from_slice(&body_vec)?;
    Ok(router_resp)
}