1 #[cfg(test)] 2 mod server_test; 3 4 pub mod config; 5 pub mod request; 6 7 use crate::allocation::allocation_manager::*; 8 use crate::auth::AuthHandler; 9 use crate::error::*; 10 use crate::proto::lifetime::DEFAULT_LIFETIME; 11 use config::*; 12 use request::*; 13 14 use std::collections::HashMap; 15 use std::sync::Arc; 16 use tokio::sync::{watch, Mutex}; 17 use tokio::time::{Duration, Instant}; 18 use util::Conn; 19 20 const INBOUND_MTU: usize = 1500; 21 22 /// Server is an instance of the TURN Server 23 pub struct Server { 24 auth_handler: Arc<dyn AuthHandler + Send + Sync>, 25 realm: String, 26 channel_bind_timeout: Duration, 27 pub(crate) nonces: Arc<Mutex<HashMap<String, Instant>>>, 28 shutdown_tx: Mutex<Option<watch::Sender<bool>>>, 29 } 30 31 impl Server { 32 /// creates the TURN server 33 pub async fn new(config: ServerConfig) -> Result<Self> { 34 config.validate()?; 35 36 let (shutdown_tx, shutdown_rx) = watch::channel(false); 37 38 let mut s = Server { 39 auth_handler: config.auth_handler, 40 realm: config.realm, 41 channel_bind_timeout: config.channel_bind_timeout, 42 nonces: Arc::new(Mutex::new(HashMap::new())), 43 shutdown_tx: Mutex::new(Some(shutdown_tx)), 44 }; 45 46 if s.channel_bind_timeout == Duration::from_secs(0) { 47 s.channel_bind_timeout = DEFAULT_LIFETIME; 48 } 49 50 for p in config.conn_configs.into_iter() { 51 let nonces = Arc::clone(&s.nonces); 52 let auth_handler = Arc::clone(&s.auth_handler); 53 let realm = s.realm.clone(); 54 let channel_bind_timeout = s.channel_bind_timeout; 55 let shutdown_rx = shutdown_rx.clone(); 56 57 tokio::spawn(async move { 58 let allocation_manager = Arc::new(Manager::new(ManagerConfig { 59 relay_addr_generator: p.relay_addr_generator, 60 })); 61 62 Server::read_loop( 63 p.conn, 64 allocation_manager, 65 nonces, 66 auth_handler, 67 realm, 68 channel_bind_timeout, 69 shutdown_rx, 70 ) 71 .await; 72 }); 73 } 74 75 Ok(s) 76 } 77 78 async fn read_loop( 79 conn: Arc<dyn Conn + Send + Sync>, 80 allocation_manager: Arc<Manager>, 81 nonces: Arc<Mutex<HashMap<String, Instant>>>, 82 auth_handler: Arc<dyn AuthHandler + Send + Sync>, 83 realm: String, 84 channel_bind_timeout: Duration, 85 mut shutdown_rx: watch::Receiver<bool>, 86 ) { 87 let mut buf = vec![0u8; INBOUND_MTU]; 88 89 loop { 90 let (n, addr) = tokio::select! { 91 v = conn.recv_from(&mut buf) => { 92 match v { 93 Ok(v) => v, 94 Err(err) => { 95 log::debug!("exit read loop on error: {}", err); 96 break; 97 } 98 } 99 }, 100 did_change = shutdown_rx.changed() => { 101 if did_change.is_err() || *shutdown_rx.borrow() { 102 // if did_change.is_err, sender was dropped, or if 103 // bool is set to true, that means we're shutting down. 104 break 105 } else { 106 continue; 107 } 108 } 109 }; 110 111 let mut r = Request { 112 conn: Arc::clone(&conn), 113 src_addr: addr, 114 buff: buf[..n].to_vec(), 115 allocation_manager: Arc::clone(&allocation_manager), 116 nonces: Arc::clone(&nonces), 117 auth_handler: Arc::clone(&auth_handler), 118 realm: realm.clone(), 119 channel_bind_timeout, 120 }; 121 122 if let Err(err) = r.handle_request().await { 123 log::error!("error when handling datagram: {}", err); 124 } 125 } 126 127 let _ = allocation_manager.close().await; 128 let _ = conn.close().await; 129 } 130 131 /// Close stops the TURN Server. It cleans up any associated state and closes all connections it is managing 132 pub async fn close(&self) -> Result<()> { 133 let mut shutdown_tx = self.shutdown_tx.lock().await; 134 if let Some(tx) = shutdown_tx.take() { 135 // errors if there are no receivers, but that's irrelevant. 136 let _ = tx.send(true); 137 // wait for all receivers to drop/close. 138 tx.closed().await; 139 } 140 141 Ok(()) 142 } 143 } 144