rlibdht 0.1.0

Rust DHT library
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

use std::sync::{Arc, Mutex};
use std::time::{SystemTime, UNIX_EPOCH};
use crate::kad::kademlia_base::KademliaBase;
use crate::messages::find_node_request::FindNodeRequest;
use crate::messages::find_node_response::FindNodeResponse;
use crate::messages::inter::message_base::MessageBase;
use crate::messages::ping_request::PingRequest;
use crate::routing::kb::k_bucket::MAX_BUCKET_SIZE;
use crate::rpc::events::error_response_event::ErrorResponseEvent;
use crate::rpc::events::inter::message_event::MessageEvent;
use crate::rpc::events::inter::response_callback::ResponseCallback;
use crate::rpc::events::response_event::ResponseEvent;
use crate::rpc::events::stalled_event::StalledEvent;
use crate::rpc::ping_response_listener::PingResponseListener;
use crate::utils::node::Node;
use crate::utils::uid::ID_LENGTH;
use super::inter::task::Task;

#[derive(Clone)]
pub struct BucketRefreshTask {
    kademlia: Box<dyn KademliaBase>
}

impl BucketRefreshTask {

    pub fn new(kademlia: &dyn KademliaBase) -> Self {
        Self {
            kademlia: kademlia.clone_dyn()
        }
    }
}

impl Task for BucketRefreshTask {

    fn execute(&self) {
        let listener = Box::new(FindNodeResponseListener::new(self.kademlia.as_ref()));
        println!("EXECUTING BUCKET REFRESH");

        for i in 1..ID_LENGTH*8 {
            if self.kademlia.get_routing_table().lock().unwrap().bucket_size(i) < MAX_BUCKET_SIZE {
                let k = self.kademlia.get_routing_table().lock().unwrap().get_derived_uid().generate_node_id_by_distance(i);

                let closest = self.kademlia.get_routing_table().lock().unwrap().find_closest(&k, MAX_BUCKET_SIZE);
                if closest.is_empty() {
                    continue;
                }

                for node in closest {
                    let mut request = FindNodeRequest::default();
                    request.set_destination(node.address);
                    request.set_target(k);

                    self.kademlia.get_server().lock().unwrap().send_with_node_callback(&mut request, node, listener.clone()).unwrap();
                }
            }
        }
    }

    fn clone_dyn(&self) -> Box<dyn Task> {
        Box::new(self.clone())
    }
}

#[derive(Clone)]
pub struct FindNodeResponseListener {
    kademlia: Box<dyn KademliaBase>,
    listener: PingResponseListener,
    queries: Arc<Mutex<Vec<Node>>> //MAY NEED TO BE RC
}

impl FindNodeResponseListener {

    pub fn new(kademlia: &dyn KademliaBase) -> Self {
        Self {
            kademlia: kademlia.clone_dyn(),
            listener: PingResponseListener::new(kademlia.get_routing_table().clone()),
            queries: Arc::new(Mutex::new(Vec::new()))
        }
    }
}

impl ResponseCallback for FindNodeResponseListener {

    fn on_response(&self, _event: ResponseEvent) {
        _event.get_node().seen();
        println!("SEEN FN {}", _event.get_node().to_string());
        let response = _event.get_message().as_any().downcast_ref::<FindNodeResponse>().unwrap();

        if response.has_nodes() {
            let mut nodes = response.get_all_nodes();

            let now = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .expect("Time went backwards")
                .as_millis();

            let uid = self.kademlia.get_routing_table().lock().unwrap().get_derived_uid();
            nodes.retain(|node| {
                if uid == node.uid ||
                        self.queries.lock().unwrap().contains(node) ||
                        self.kademlia.get_routing_table().lock().unwrap().has_queried(node, now) {
                    false

                } else {
                    true
                }
            });

            for node in &nodes {
                self.queries.lock().unwrap().push(node.clone());
            }

            for node in &nodes {
                if self.kademlia.get_routing_table().lock().unwrap().is_secure_only() && !node.has_secure_id() {
                    println!("SKIPPING {}  {}  {}", now, node.last_seen, node.to_string());
                    continue;
                }

                let mut req = PingRequest::default();
                req.set_destination(node.address);
                self.kademlia.get_server().lock().unwrap().send_with_node_callback(&mut req, node.clone(), Box::new(self.listener.clone())).unwrap();
            }
        }
    }

    fn on_error_response(&self, _event: ErrorResponseEvent) {
        _event.get_node().seen();
    }

    fn on_stalled(&self, _event: StalledEvent) {
        _event.get_node().mark_stale();
    }
}