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

use std::{collections::HashMap, sync::Arc, time::Duration};

use actix::prelude::*;

use super::model::{
    NamingRouteRequest, NamingRouterResponse, SyncSenderRequest, SyncSenderResponse,
    SyncSenderSetCmd,
};
use crate::common::constant::{GRPC_HEAD_KEY_CLUSTER_ID, GRPC_HEAD_KEY_SUB_NAME};
use crate::grpc::handler::NAMING_ROUTE_REQUEST;
use crate::{grpc::PayloadUtils, raft::network::factory::RaftClusterRequestSender};

pub struct ClusteSyncSender {
    //local_id:u64,
    target_id: u64,
    target_addr: Arc<String>,
    pub(crate) cluster_sender: Option<Arc<RaftClusterRequestSender>>,
    send_extend_infos: HashMap<String, String>,
}

impl ClusteSyncSender {
    pub fn new(
        local_id: u64,
        target_id: u64,
        target_addr: Arc<String>,
        cluster_sender: Option<Arc<RaftClusterRequestSender>>,
    ) -> Self {
        let mut send_extend_infos = HashMap::default();
        send_extend_infos.insert(GRPC_HEAD_KEY_CLUSTER_ID.to_owned(), local_id.to_string());
        Self {
            //local_id,
            target_id,
            target_addr,
            cluster_sender,
            send_extend_infos,
        }
    }
}

impl Actor for ClusteSyncSender {
    type Context = Context<Self>;

    fn started(&mut self, _ctx: &mut Self::Context) {
        log::info!("ClusteSyncSender started,target_id:{}", &self.target_id);
    }
}

impl Handler<SyncSenderSetCmd> for ClusteSyncSender {
    type Result = anyhow::Result<SyncSenderResponse>;

    fn handle(&mut self, msg: SyncSenderSetCmd, _ctx: &mut Self::Context) -> Self::Result {
        match msg {
            SyncSenderSetCmd::UpdateTargetAddr(target_addr) => {
                self.target_addr = target_addr;
                Ok(SyncSenderResponse::None)
            }
        }
    }
}

impl Handler<SyncSenderRequest> for ClusteSyncSender {
    type Result = ResponseActFuture<Self, anyhow::Result<SyncSenderResponse>>;

    fn handle(&mut self, msg: SyncSenderRequest, _ctx: &mut Self::Context) -> Self::Result {
        let cluster_sender = self.cluster_sender.clone();
        let target_addr = self.target_addr.clone();
        let mut send_extend_infos = self.send_extend_infos.clone();
        send_extend_infos.insert(
            GRPC_HEAD_KEY_SUB_NAME.to_string(),
            msg.0.get_sub_name().to_string(),
        );
        let fut = async move {
            let cluster_sender = match cluster_sender {
                Some(v) => v,
                None => {
                    return Err(anyhow::anyhow!(
                        "ClusteSyncSender,the cluster sender is none!"
                    ))
                }
            };
            let req = msg.0;
            let request = serde_json::to_string(&req).unwrap_or_default();
            let payload = PayloadUtils::build_full_payload(
                NAMING_ROUTE_REQUEST,
                request,
                "",
                send_extend_infos.clone(),
            );
            let resp_payload = match cluster_sender
                .send_request(target_addr.clone(), payload)
                .await
            {
                Ok(v) => v,
                Err(err) => {
                    if let NamingRouteRequest::Ping(_) = &req {
                        //ping 不重试
                        return Err(err);
                    }
                    //retry request
                    tokio::time::sleep(Duration::from_millis(100)).await;
                    let request = serde_json::to_string(&req).unwrap_or_default();
                    let payload = PayloadUtils::build_full_payload(
                        NAMING_ROUTE_REQUEST,
                        request,
                        "",
                        send_extend_infos,
                    );
                    cluster_sender.send_request(target_addr, payload).await?
                }
            };
            let body_vec = resp_payload.body.unwrap_or_default().value;
            let _: NamingRouterResponse = serde_json::from_slice(&body_vec)?;
            Ok(SyncSenderResponse::None)
        }
        .into_actor(self)
        .map(|r, _act, _ctx| r);
        Box::pin(fut)
    }
}