volans_bridge/backend/
handler.rs

1use std::{
2    convert::Infallible,
3    fmt,
4    task::{Context, Poll},
5    time::Duration,
6};
7
8use futures::FutureExt;
9use futures_bounded::FuturesSet;
10use volans_core::{Multiaddr, PeerId, upgrade::ReadyUpgrade};
11use volans_swarm::{
12    ConnectionHandler, ConnectionHandlerEvent, InboundStreamHandler, InboundUpgradeSend,
13    StreamProtocol, SubstreamProtocol,
14};
15
16use crate::protocol;
17
18/// 后端处理代理协议
19pub struct Handler {
20    relay_remote_addr: Multiaddr,
21    inbound_pending_circuits: FuturesSet<Result<protocol::Relay, protocol::Error>>,
22}
23
24impl Handler {
25    pub fn new(relay_remote_addr: Multiaddr) -> Self {
26        Self {
27            relay_remote_addr,
28            inbound_pending_circuits: FuturesSet::new(
29                || futures_bounded::Delay::futures_timer(Duration::from_secs(15)),
30                10, // 最大并行处理数
31            ),
32        }
33    }
34}
35
36impl ConnectionHandler for Handler {
37    type Action = Infallible;
38    type Event = NewCircuitAccept;
39
40    fn handle_action(&mut self, _action: Self::Action) {
41        // No actions to handle
42    }
43
44    fn poll(&mut self, cx: &mut Context<'_>) -> Poll<ConnectionHandlerEvent<Self::Event>> {
45        loop {
46            match self.inbound_pending_circuits.poll_unpin(cx) {
47                Poll::Ready(Ok(Ok(protocol::Relay {
48                    circuit,
49                    src_peer_id,
50                    dst_peer_id,
51                    src_relayed_addr,
52                }))) => {
53                    tracing::info!("Inbound circuit request accepted");
54                    let event = NewCircuitAccept {
55                        relay_remote_addr: self.relay_remote_addr.clone(),
56                        circuit,
57                        src_peer_id,
58                        dst_peer_id,
59                        src_relayed_addr,
60                    };
61                    return Poll::Ready(ConnectionHandlerEvent::Notify(event));
62                }
63                Poll::Ready(Ok(Err(e))) => {
64                    tracing::warn!("Inbound circuit error: {:?}", e);
65                    continue;
66                }
67                Poll::Ready(Err(e)) => {
68                    tracing::error!("Inbound circuit processing failed: {:?}", e);
69                    continue;
70                }
71                Poll::Pending => {}
72            }
73            return Poll::Pending;
74        }
75    }
76
77    fn poll_close(&mut self, _cx: &mut Context<'_>) -> Poll<Option<Self::Event>> {
78        Poll::Ready(None)
79    }
80}
81
82impl InboundStreamHandler for Handler {
83    type InboundUpgrade = ReadyUpgrade<StreamProtocol>;
84    type InboundUserData = ();
85
86    fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundUpgrade, Self::InboundUserData> {
87        SubstreamProtocol::new(ReadyUpgrade::new(protocol::PROTOCOL_NAME), ())
88    }
89
90    fn on_fully_negotiated(
91        &mut self,
92        _user_data: Self::InboundUserData,
93        stream: <Self::InboundUpgrade as InboundUpgradeSend>::Output,
94    ) {
95        let result = self
96            .inbound_pending_circuits
97            .try_push(protocol::handle_bridge_relay_connect(stream).boxed());
98        if result.is_err() {
99            tracing::warn!("Failed to push inbound circuit");
100        }
101    }
102
103    fn on_upgrade_error(
104        &mut self,
105        _user_data: Self::InboundUserData,
106        _error: <Self::InboundUpgrade as InboundUpgradeSend>::Error,
107    ) {
108        tracing::warn!("New circuit accept eRR");
109    }
110}
111
112pub struct NewCircuitAccept {
113    pub(crate) relay_remote_addr: Multiaddr,
114    pub(crate) circuit: protocol::Circuit,
115    pub(crate) src_peer_id: PeerId,
116    pub(crate) dst_peer_id: PeerId,
117    pub(crate) src_relayed_addr: Multiaddr,
118}
119
120impl fmt::Debug for NewCircuitAccept {
121    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
122        f.debug_struct("NewCircuitAccept")
123            .field("src_peer_id", &self.src_peer_id)
124            .field("dst_peer_id", &self.dst_peer_id)
125            .field("src_relayed_addr", &self.src_relayed_addr)
126            .finish()
127    }
128}
volans_bridge/backend/handler.rs

volans_bridge/backend/
handler.rs
