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
use crate::{protocols::ReturnableItem, Pea2Pea};
use tokio::{
sync::{mpsc, oneshot},
task,
};
use tracing::*;
use std::net::SocketAddr;
#[async_trait::async_trait]
pub trait Disconnect: Pea2Pea
where
Self: Clone + Send + Sync + 'static,
{
fn enable_disconnect(&self) {
let (from_node_sender, mut from_node_receiver) =
mpsc::channel::<(SocketAddr, oneshot::Sender<()>)>(
self.node().config().protocol_handler_queue_depth,
);
let self_clone = self.clone();
let disconnect_task = tokio::spawn(async move {
trace!(parent: self_clone.node().span(), "spawned the Disconnect handler task");
while let Some((addr, notifier)) = from_node_receiver.recv().await {
let self_clone2 = self_clone.clone();
task::spawn(async move {
self_clone2.handle_disconnect(addr).await;
let _ = notifier.send(());
});
}
});
self.node().tasks.lock().push(disconnect_task);
let hdl = DisconnectHandler(from_node_sender);
assert!(
self.node().protocols.disconnect_handler.set(hdl).is_ok(),
"the Disconnect protocol was enabled more than once!"
);
}
async fn handle_disconnect(&self, addr: SocketAddr);
}
pub struct DisconnectHandler(mpsc::Sender<ReturnableItem<SocketAddr, ()>>);
impl DisconnectHandler {
pub(crate) async fn trigger(&self, item: ReturnableItem<SocketAddr, ()>) {
if self.0.send(item).await.is_err() {
unreachable!();
}
}
}