-
Notifications
You must be signed in to change notification settings - Fork 222
/
ncp.rs
86 lines (77 loc) · 2.5 KB
/
ncp.rs
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
use blockthread::BlockThread;
use service::Service;
use std::net::UdpSocket;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::channel;
use std::sync::{Arc, RwLock};
use std::thread::{self, JoinHandle};
use streamer;
use window::SharedWindow;
pub struct Ncp {
exit: Arc<AtomicBool>,
thread_hdls: Vec<JoinHandle<()>>,
}
impl Ncp {
pub fn new(
blockthread: &Arc<RwLock<BlockThread>>,
window: SharedWindow,
ledger_path: Option<&str>,
gossip_socket: UdpSocket,
exit: Arc<AtomicBool>,
) -> Self {
let (request_sender, request_receiver) = channel();
let gossip_socket = Arc::new(gossip_socket);
trace!(
"Ncp: id: {:?}, listening on: {:?}",
&blockthread.read().unwrap().id.as_ref()[..4],
gossip_socket.local_addr().unwrap()
);
let t_receiver =
streamer::blob_receiver(gossip_socket.clone(), exit.clone(), request_sender);
let (response_sender, response_receiver) = channel();
let t_responder = streamer::responder("ncp", gossip_socket, response_receiver);
let t_listen = BlockThread::listen(
blockthread.clone(),
window,
ledger_path,
request_receiver,
response_sender.clone(),
exit.clone(),
);
let t_gossip = BlockThread::gossip(blockthread.clone(), response_sender, exit.clone());
let thread_hdls = vec![t_receiver, t_responder, t_listen, t_gossip];
Ncp { exit, thread_hdls }
}
pub fn close(self) -> thread::Result<()> {
self.exit.store(true, Ordering::Relaxed);
self.join()
}
}
impl Service for Ncp {
type JoinReturnType = ();
fn join(self) -> thread::Result<()> {
for thread_hdl in self.thread_hdls {
thread_hdl.join()?;
}
Ok(())
}
}
#[cfg(test)]
mod tests {
use blockthread::{BlockThread, Node};
use ncp::Ncp;
use std::sync::atomic::AtomicBool;
use std::sync::{Arc, RwLock};
#[test]
#[ignore]
// test that stage will exit when flag is set
fn test_exit() {
let exit = Arc::new(AtomicBool::new(false));
let tn = Node::new_localhost();
let blockthread = BlockThread::new(tn.info.clone()).expect("BlockThread::new");
let c = Arc::new(RwLock::new(blockthread));
let w = Arc::new(RwLock::new(vec![]));
let d = Ncp::new(&c, w, None, tn.sockets.gossip, exit.clone());
d.close().expect("thread join");
}
}