v_common_search/
ft_client.rs

1use crate::common::{FTQuery, QueryResult};
2use nng::{Message, Protocol, Socket};
3use serde_json::Value;
4use std::{thread, time};
5use v_api::app::ResultCode;
6use std::time::Duration;
7use nng::options::{RecvTimeout, Options, SendTimeout};
8
9pub struct FTClient {
10    client: Socket,
11    addr: String,
12    is_ready: bool,
13}
14
15impl FTClient {
16    pub fn new(_ro_client_addr: String) -> FTClient {
17        FTClient {
18            client: Socket::new(Protocol::Req0).unwrap(),
19            addr: _ro_client_addr,
20            is_ready: false,
21        }
22    }
23
24    pub fn connect(&mut self) -> bool {
25        if let Err(e) = self.client.dial(self.addr.as_str()) {
26            error!("ft-client:fail dial to ft-service, [{}], err={}", self.addr, e);
27        } else {
28            info!("success connect to ft-service, [{}]", self.addr);
29
30            if let Err(e) = self.client.set_opt::<RecvTimeout>(Some(Duration::from_secs(30))) {
31                error!("fail set recv timeout, err={}", e);
32            }
33            if let Err(e) = self.client.set_opt::<SendTimeout>(Some(Duration::from_secs(30))) {
34                error!("fail set send timeout, err={}", e);
35            }
36
37            self.is_ready = true;
38        }
39        self.is_ready
40    }
41
42    pub fn query(&mut self, query: FTQuery) -> QueryResult {
43        let mut res = QueryResult::default();
44
45        if !self.is_ready {
46            while !self.connect() {
47                error!("not ready, sleep...");
48                thread::sleep(time::Duration::from_millis(3000));
49            }
50        }
51
52        if !self.is_ready {
53            res.result_code = ResultCode::NotReady;
54            return res;
55        }
56
57        let req = Message::from(query.as_string().as_bytes());
58
59        if let Err(e) = self.client.send(req) {
60            error!("fail send to search module, err={:?}", e);
61            res.result_code = ResultCode::NotReady;
62            return res;
63        }
64
65        // Wait for the response from the server.
66        let wmsg = self.client.recv();
67
68        if let Err(e) = wmsg {
69            error!("fail recv from search module, err={:?}", e);
70            res.result_code = ResultCode::NotReady;
71            return res;
72        }
73
74        let msg = wmsg.unwrap();
75
76        let reply = String::from_utf8_lossy(&msg);
77
78        let v: Value = if let Ok(v) = serde_json::from_str(&reply) {
79            v
80        } else {
81            Value::Null
82        };
83
84        res.result_code = ResultCode::from_i64(v["result_code"].as_i64().unwrap_or_default());
85
86        if res.result_code == ResultCode::Ok {
87            let jarray: &Vec<_> = &v["result"].as_array().expect("array");
88            res.result = jarray.iter().map(|v| v.as_str().unwrap_or_default().to_owned()).collect();
89
90            res.count = v["count"].as_i64().unwrap_or_default();
91            res.estimated = v["estimated"].as_i64().unwrap_or_default();
92            res.processed = v["processed"].as_i64().unwrap_or_default();
93            res.cursor = v["cursor"].as_i64().unwrap_or_default();
94        }
95
96        //info!("msg={}", v);
97        res
98    }
99}
v_common_search/ft_client.rs

v_common_search/
ft_client.rs
