1 use crate::mux::mux_func::MatchFunc; 2 use util::{Buffer, Conn}; 3 4 use async_trait::async_trait; 5 use std::collections::HashMap; 6 use std::io; 7 use std::net::SocketAddr; 8 use std::sync::Arc; 9 use tokio::sync::Mutex; 10 11 /// Endpoint implements net.Conn. It is used to read muxed packets. 12 pub struct Endpoint { 13 pub(crate) id: usize, 14 pub(crate) buffer: Buffer, 15 pub(crate) match_fn: MatchFunc, 16 pub(crate) next_conn: Arc<dyn Conn + Send + Sync>, 17 pub(crate) endpoints: Arc<Mutex<HashMap<usize, Arc<Endpoint>>>>, 18 } 19 20 impl Endpoint { 21 /// Close unregisters the endpoint from the Mux close(&self) -> Result<()>22 pub async fn close(&self) -> Result<()> { 23 self.buffer.close().await; 24 25 let mut endpoints = self.endpoints.lock().await; 26 endpoints.remove(&self.id); 27 28 Ok(()) 29 } 30 } 31 32 type Result<T> = std::result::Result<T, util::Error>; 33 34 #[async_trait] 35 impl Conn for Endpoint { connect(&self, _addr: SocketAddr) -> Result<()>36 async fn connect(&self, _addr: SocketAddr) -> Result<()> { 37 Err(io::Error::new(io::ErrorKind::Other, "Not applicable").into()) 38 } 39 40 /// reads a packet of len(p) bytes from the underlying conn 41 /// that are matched by the associated MuxFunc recv(&self, buf: &mut [u8]) -> Result<usize>42 async fn recv(&self, buf: &mut [u8]) -> Result<usize> { 43 match self.buffer.read(buf, None).await { 44 Ok(n) => Ok(n), 45 Err(err) => Err(io::Error::new(io::ErrorKind::Other, err.to_string()).into()), 46 } 47 } recv_from(&self, _buf: &mut [u8]) -> Result<(usize, SocketAddr)>48 async fn recv_from(&self, _buf: &mut [u8]) -> Result<(usize, SocketAddr)> { 49 Err(io::Error::new(io::ErrorKind::Other, "Not applicable").into()) 50 } 51 52 /// writes bytes to the underlying conn send(&self, buf: &[u8]) -> Result<usize>53 async fn send(&self, buf: &[u8]) -> Result<usize> { 54 self.next_conn.send(buf).await 55 } 56 send_to(&self, _buf: &[u8], _target: SocketAddr) -> Result<usize>57 async fn send_to(&self, _buf: &[u8], _target: SocketAddr) -> Result<usize> { 58 Err(io::Error::new(io::ErrorKind::Other, "Not applicable").into()) 59 } 60 local_addr(&self) -> Result<SocketAddr>61 fn local_addr(&self) -> Result<SocketAddr> { 62 self.next_conn.local_addr() 63 } 64 remote_addr(&self) -> Option<SocketAddr>65 fn remote_addr(&self) -> Option<SocketAddr> { 66 self.next_conn.remote_addr() 67 } 68 close(&self) -> Result<()>69 async fn close(&self) -> Result<()> { 70 self.next_conn.close().await 71 } 72 } 73