smol -> tokio

src/webrtc.rs

@@ -1,7 +1,7 @@
 use dashmap::DashMap;
-use smol::{
-    channel::{Receiver, Sender},
+use tokio::{
     net::UdpSocket,
+    sync::mpsc::{Receiver, Sender},
 };
 use std::{
     error::Error,
@@ -10,7 +10,7 @@ use std::{
 };
 
 use str0m::{
-    Candidate, Event, IceConnectionState, Input, Output, Rtc,
+    Candidate, Event, Input, Output, Rtc,
     change::SdpOffer,
     media::{MediaKind, MediaTime, Mid},
     net::{Protocol, Receive},
@@ -25,9 +25,9 @@ pub struct Webrtc {
 }
 
 impl Webrtc {
-    pub fn start(self) -> Result<(), Box<dyn Error>> {
-        smol::spawn(async move {
-            while let Ok(offer) = self.offer_rx.recv().await {
+    pub fn start(mut self) -> Result<(), Box<dyn Error>> {
+        tokio::spawn(async move {
+            while let Some(offer) = self.offer_rx.recv().await {
                 let (stream_key, sdp_body) = offer;
                 let socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
                 let local_addr = socket.local_addr().unwrap();
@@ -69,13 +69,11 @@ impl Webrtc {
                     .unwrap();
 
                 let sessions_ref = self.sessions_ref.clone();
-                smol::spawn(async move {
+                tokio::spawn(async move {
                     Webrtc::detach_connection(socket, rtc, sessions_ref, mid).await;
-                })
-                .detach();
+                });
             }
-        })
-        .detach();
+        });
         Ok(())
     }
 
@@ -139,13 +137,10 @@ impl Webrtc {
                     }
                 }
                 if let Some(ref mut stream) = video_stream {
-                    // Drain at most 8 frames per loop tick so the UDP socket
-                    // (ICE keepalives, RTCP) is not starved by a backlog.
                     for _ in 0..8 {
                         match stream.try_recv() {
                             Ok(frame) => {
                                 let now = Instant::now();
-                                // Explicit 90 kHz clock for H.264 RTP timestamps.
                                 let rtp_time = MediaTime::from_90khz(frame.timestamp_ms as u64 * 90);
                                 if let (Some(pt), Some(writer)) =
                                     (video_pt, video_mid.and_then(|m| rtc.writer(m)))
@@ -159,48 +154,34 @@ impl Webrtc {
                             }
                             Err(async_broadcast::TryRecvError::Empty) => break,
                             Err(async_broadcast::TryRecvError::Closed) => return,
-                            // Overflow means some frames were dropped; the next
-                            // try_recv will give the oldest surviving frame, so
-                            // continue draining rather than breaking.
                             Err(async_broadcast::TryRecvError::Overflowed(_)) => continue,
                         }
                     }
                 }
             }
 
-            // Cap wait to 20 ms so frame delivery stays timely even when
-            // str0m's deadline is far out.
             let wait_until = deadline.min(Instant::now() + Duration::from_millis(20)).max(Instant::now());
+            let sleep = tokio::time::sleep_until(wait_until.into());
 
-            let input = smol::future::or(
-                async {
-                    smol::Timer::at(wait_until).await;
-                    None
-                },
-                async {
-                    let (n, from) = socket.recv_from(&mut recv_buf).await.ok()?;
-                    Some((n, from))
-                },
-            )
-            .await;
-
-            match input {
-                None => {
+            tokio::select! {
+                _ = sleep => {
                     rtc.handle_input(Input::Timeout(Instant::now())).ok();
                 }
-                Some((n, from)) => {
-                    let data = recv_buf[..n].to_vec();
-                    if let Ok(contents) = data.as_slice().try_into() {
-                        rtc.handle_input(Input::Receive(
-                            Instant::now(),
-                            Receive {
-                                proto: Protocol::Udp,
-                                source: from,
-                                destination: local_addr,
-                                contents,
-                            },
-                        ))
-                        .ok();
+                result = socket.recv_from(&mut recv_buf) => {
+                    if let Ok((n, from)) = result {
+                        let data = recv_buf[..n].to_vec();
+                        if let Ok(contents) = data.as_slice().try_into() {
+                            rtc.handle_input(Input::Receive(
+                                Instant::now(),
+                                Receive {
+                                    proto: Protocol::Udp,
+                                    source: from,
+                                    destination: local_addr,
+                                    contents,
+                                },
+                            ))
+                            .ok();
+                        }
                     }
                 }
             }