volans_bridge/backend/
behavior.rs

1use std::{
2    convert::Infallible,
3    task::{Context, Poll},
4};
5
6use futures::{StreamExt, channel::mpsc};
7use volans_core::{Multiaddr, PeerId, multiaddr::Protocol};
8use volans_swarm::{
9    BehaviorEvent, ConnectionDenied, ConnectionId, NetworkBehavior, NetworkIncomingBehavior,
10    THandlerAction, THandlerEvent,
11};
12
13use crate::transport::{Connection, IncomingRelayedConnection, TransportRequest};
14
15use super::handler;
16
17pub struct Behavior {
18    transport_request_receiver: mpsc::Receiver<TransportRequest>,
19    listener: Option<mpsc::Sender<IncomingRelayedConnection>>,
20}
21
22impl Behavior {
23    pub fn new(transport_request_receiver: mpsc::Receiver<TransportRequest>) -> Self {
24        Self {
25            transport_request_receiver,
26            listener: None,
27        }
28    }
29}
30
31impl NetworkBehavior for Behavior {
32    type ConnectionHandler = handler::Handler;
33    type Event = Infallible;
34
35    fn on_connection_handler_event(
36        &mut self,
37        _id: ConnectionId,
38        peer_id: PeerId,
39        handler::NewCircuitAccept {
40            relay_remote_addr,
41            circuit,
42            src_peer_id,
43            dst_peer_id: _,
44            src_relayed_addr,
45        }: THandlerEvent<Self>,
46    ) {
47        match self.listener {
48            Some(ref mut sender) => {
49                let r = sender.try_send(IncomingRelayedConnection::new(
50                    Connection::new_accepting(circuit),
51                    src_peer_id,
52                    peer_id,
53                    src_relayed_addr,
54                ));
55                if let Err(e) = r {
56                    tracing::error!("Failed to send incoming relayed connection: {}", e);
57                }
58            }
59            None => {
60                tracing::warn!(
61                    "No listener found for remote address: {}",
62                    relay_remote_addr
63                );
64            }
65        }
66    }
67
68    fn poll(
69        &mut self,
70        cx: &mut Context<'_>,
71    ) -> Poll<BehaviorEvent<Self::Event, THandlerAction<Self>>> {
72        loop {
73            match self.transport_request_receiver.poll_next_unpin(cx) {
74                Poll::Ready(Some(TransportRequest::ListenRequest {
75                    local_addr,
76                    listener_sender,
77                })) => {
78                    tracing::debug!("Circuit Listening on: {:?}", local_addr);
79                    self.listener = Some(listener_sender);
80                    continue;
81                }
82                Poll::Ready(Some(TransportRequest::DialRequest { .. })) => {
83                    tracing::error!("Unexpected DialRequest in Backend Behavior");
84                }
85                Poll::Ready(None) => {}
86                Poll::Pending => {}
87            }
88            return Poll::Pending;
89        }
90    }
91}
92
93impl NetworkIncomingBehavior for Behavior {
94    /// 处理已建立的连接
95    fn handle_established_connection(
96        &mut self,
97        _id: ConnectionId,
98        peer_id: PeerId,
99        _local_addr: &Multiaddr,
100        remote_addr: &Multiaddr,
101    ) -> Result<Self::ConnectionHandler, ConnectionDenied> {
102        let relay_addr = remote_addr.clone().with(Protocol::Peer(peer_id));
103        Ok(handler::Handler::new(relay_addr))
104    }
105}
volans_bridge/backend/behavior.rs

volans_bridge/backend/
behavior.rs
