volans_bridge/relay/server/
behavior.rs

1use std::{
2    collections::VecDeque,
3    convert::Infallible,
4    task::{Context, Poll},
5};
6
7use futures::channel::mpsc;
8use volans_core::{Multiaddr, PeerId, multiaddr::Protocol};
9use volans_swarm::{
10    BehaviorEvent, ConnectionDenied, ConnectionId, NetworkBehavior, NetworkIncomingBehavior,
11    THandlerAction, THandlerEvent,
12};
13
14use crate::relay::CircuitRequest;
15
16use super::handler;
17
18pub struct Behavior {
19    local_peer_id: PeerId,
20    pending_requests: VecDeque<CircuitRequest>,
21    request_sender: mpsc::UnboundedSender<CircuitRequest>,
22}
23
24impl Behavior {
25    pub fn new(
26        local_peer_id: PeerId,
27        request_sender: mpsc::UnboundedSender<CircuitRequest>,
28    ) -> Self {
29        Self {
30            local_peer_id,
31            pending_requests: VecDeque::new(),
32            request_sender,
33        }
34    }
35}
36
37impl NetworkBehavior for Behavior {
38    type ConnectionHandler = handler::Handler;
39    type Event = Infallible;
40
41    fn on_connection_handler_event(
42        &mut self,
43        id: ConnectionId,
44        peer_id: PeerId,
45        handler::CircuitAccepted {
46            dst_peer_id,
47            dst_addresses,
48            circuit,
49            relayed_addr,
50        }: THandlerEvent<Self>,
51    ) {
52        // 客户端发起一个中继请求,
53        let request = CircuitRequest {
54            relayed_addr,
55            dst_peer_id,
56            dst_addresses,
57            src_peer_id: peer_id,
58            src_connection_id: id,
59            circuit,
60        };
61        // 写入待处理请求队列
62        self.pending_requests.push_back(request);
63    }
64
65    fn poll(
66        &mut self,
67        _cx: &mut Context<'_>,
68    ) -> Poll<BehaviorEvent<Self::Event, THandlerAction<Self>>> {
69        loop {
70            if let Some(request) = self.pending_requests.pop_front() {
71                // 发送请求给客户端
72                tracing::debug!("Sending request: {:?}", request);
73                if let Err(err) = self.request_sender.unbounded_send(request) {
74                    tracing::warn!("Failed to send request: {:?}", err);
75                }
76                continue;
77            }
78            return Poll::Pending;
79        }
80    }
81}
82
83impl NetworkIncomingBehavior for Behavior {
84    /// 处理已建立的连接
85    fn handle_established_connection(
86        &mut self,
87        _id: ConnectionId,
88        peer_id: PeerId,
89        local_addr: &Multiaddr,
90        _remote_addr: &Multiaddr,
91    ) -> Result<Self::ConnectionHandler, ConnectionDenied> {
92        let relay_addr = local_addr
93            .clone()
94            .with(Protocol::Peer(self.local_peer_id))
95            .with(Protocol::Circuit)
96            .with(Protocol::Peer(peer_id));
97
98        Ok(handler::Handler::new(relay_addr))
99    }
100}
volans_bridge/relay/server/behavior.rs

volans_bridge/relay/server/
behavior.rs
