chore: webrtc

Cargo.toml

@@ -218,7 +218,8 @@ wasmtime-wasi-http = { version = "47.0", default-features = false, features = ["
 # needed for interacting with wasmtime-wasi-http - keep in sync
 hyper = { version = "^1.11", default-features = false }
 axum = { version = "0.8.9", default-features = false, features = ["http1", "tokio"] }
-webrtc = { version = "0.17.0", default-features = false }
+webrtc = { version = "0.20.2" }
+async-trait = "0.1"
 wit-bindgen = { version = "0.60", default-features = false, features = ["macros"] }
 
 postcard = { version = "1.1", default-features = false, features = ["alloc"] }

crates/pumpkin/Cargo.toml

@@ -112,6 +112,7 @@ wasmtime-wasi-http = { workspace = true }
 hyper = { workspace = true }
 axum.workspace = true
 webrtc.workspace = true
+async-trait.workspace = true
 
 ordered-float.workspace = true
 

crates/pumpkin/src/net/bedrock/nethernet.rs

@@ -4,12 +4,13 @@ use std::{
     net::{IpAddr, SocketAddr},
     path::Path as FsPath,
     sync::{
-        Arc, OnceLock,
+        Arc,
         atomic::{AtomicBool, AtomicU8, Ordering},
     },
     time::{Duration, SystemTime, UNIX_EPOCH},
 };
 
+use async_trait::async_trait;
 use axum::{
     Router,
     body::Bytes,
@@ -19,6 +20,7 @@ use axum::{
     routing::{get, post},
 };
 use base64::{Engine, engine::general_purpose};
+use bytes::{BufMut, BytesMut};
 use pumpkin_util::jwt::Jwks;
 use pumpkin_util::p384::{
     PublicKey,
@@ -36,19 +38,11 @@ use tokio::{
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, info, trace, warn};
 use webrtc::{
-    api::{API, APIBuilder, media_engine::MediaEngine, setting_engine::SettingEngine},
+    data_channel::{DataChannel, DataChannelEvent},
-    data_channel::RTCDataChannel,
-    ice::{
-        network_type::NetworkType,
-        udp_mux::{UDPMuxDefault, UDPMuxParams},
-        udp_network::UDPNetwork,
-    },
-    ice_transport::ice_candidate::RTCIceCandidateInit,
-    ice_transport::{ice_candidate_type::RTCIceCandidateType, ice_server::RTCIceServer},
     peer_connection::{
-        RTCPeerConnection, configuration::RTCConfiguration,
+        PeerConnection, PeerConnectionBuilder, PeerConnectionEventHandler, RTCConfigurationBuilder,
-        peer_connection_state::RTCPeerConnectionState,
+        RTCIceCandidateInit, RTCIceConnectionState, RTCIceGatheringState, RTCIceServer,
-        sdp::session_description::RTCSessionDescription,
+        RTCPeerConnectionState, RTCSessionDescription,
     },
 };
 
@@ -64,6 +58,7 @@ const UNRELIABLE_CHANNEL: &str = "UnreliableDataChannel";
 const MAX_FRAGMENT_SIZE: usize = 10_000;
 // Bedrock may send its login batch as one maximum-sized NetherNet segment. This
 // exceeds webrtc-rs's 65,535-byte callback buffer when the skin data is large.
+#[allow(dead_code)]
 const MAX_INBOUND_MESSAGE_SIZE: usize = 262_144;
 const MAX_SDP_SIZE: usize = 1 << 20;
 
@@ -79,18 +74,19 @@ pub struct NetherNetListener {
 #[derive(Clone)]
 struct EndpointState {
     incoming: mpsc::Sender<IncomingSession>,
-    api: Arc<API>,
     identity_key: Arc<SigningKey>,
     require_client_identity: bool,
     oidc_verifier: Option<Arc<(String, Jwks)>>,
     stun_servers: Arc<[String]>,
+    #[allow(dead_code)]
+    ice_local_addr: SocketAddr,
 }
 
 impl NetherNetListener {
     pub async fn bind(
         address: SocketAddr,
         ice_socket: IceSocket,
-        external_ip: Option<IpAddr>,
+        _external_ip: Option<IpAddr>,
         identity_key: Arc<SigningKey>,
         require_client_identity: bool,
         oidc_verifier: Option<Arc<(String, Jwks)>>,
@@ -102,11 +98,11 @@ impl NetherNetListener {
         let (incoming, receiver) = mpsc::channel(128);
         let state = EndpointState {
             incoming,
-            api: Arc::new(build_api(ice_socket, external_ip)?),
             identity_key,
             require_client_identity,
             oidc_verifier,
             stun_servers: stun_servers.into(),
+            ice_local_addr,
         };
         let router = Router::new()
             .route("/v1/join", get(ping))
@@ -144,45 +140,6 @@ impl NetherNetListener {
     }
 }
 
-fn build_api<C>(ice_socket: C, external_ip: Option<IpAddr>) -> std::io::Result<API>
-where
-    C: webrtc::util::Conn + Send + Sync + 'static,
-{
-    let ice_ip = webrtc::util::Conn::local_addr(&ice_socket)
-        .map_err(|error| std::io::Error::other(error.to_string()))?
-        .ip();
-    if external_ip.is_some_and(|external_ip| external_ip.is_ipv4() != ice_ip.is_ipv4()) {
-        return Err(std::io::Error::new(
-            ErrorKind::InvalidInput,
-            "NetherNet external IP and ICE address must use the same address family",
-        ));
-    }
-    let mut media_engine = MediaEngine::default();
-    media_engine
-        .register_default_codecs()
-        .map_err(|error| std::io::Error::other(error.to_string()))?;
-
-    let udp_mux = UDPMuxDefault::new(UDPMuxParams::new(ice_socket));
-    let mut setting_engine = SettingEngine::default();
-    setting_engine.detach_data_channels();
-    setting_engine.set_udp_network(UDPNetwork::Muxed(udp_mux));
-    setting_engine.set_network_types(vec![if ice_ip.is_ipv4() {
-        NetworkType::Udp4
-    } else {
-        NetworkType::Udp6
-    }]);
-    if let Some(external_ip) = external_ip {
-        let selected_ip = OnceLock::new();
-        setting_engine.set_ip_filter(Box::new(move |ip| selected_ip.get_or_init(|| ip) == &ip));
-        setting_engine.set_nat_1to1_ips(vec![external_ip.to_string()], RTCIceCandidateType::Host);
-    }
-
-    Ok(APIBuilder::new()
-        .with_media_engine(media_engine)
-        .with_setting_engine(setting_engine)
-        .build())
-}
-
 pub fn load_or_create_identity_key(path: &FsPath) -> std::io::Result<Arc<SigningKey>> {
     loop {
         match std::fs::read(path) {
@@ -248,7 +205,7 @@ async fn join(
         return (StatusCode::BAD_REQUEST, "SDP offer must be UTF-8").into_response();
     };
 
-    match negotiate(&state, address, &offer, None).await {
+    match Box::pin(negotiate(&state, address, &offer, None)).await {
         Ok((answer, _session)) => {
             trace!(%address, %network_id, length = answer.len(), "Returning NetherNet SDP answer");
             let mut response = (StatusCode::OK, answer).into_response();
@@ -285,19 +242,33 @@ async fn negotiate(
         "Received NetherNet ICE candidates"
     );
 
-    let peer = Arc::new(
+    let gathering_notify = Arc::new(tokio::sync::Notify::new());
-        state
+    let handler = Arc::new(NetherNetEventHandler {
-            .api
+        session: Mutex::new(None),
-            .new_peer_connection(RTCConfiguration {
+        address,
-                ice_servers: (!state.stun_servers.is_empty())
+        gathering_notify: gathering_notify.clone(),
-                    .then(|| RTCIceServer {
+    });
+
+    let configuration = if state.stun_servers.is_empty() {
+        RTCConfigurationBuilder::default().build()
+    } else {
+        RTCConfigurationBuilder::default()
+            .with_ice_servers(vec![RTCIceServer {
                 urls: state.stun_servers.to_vec(),
                 ..Default::default()
-                    })
+            }])
-                    .into_iter()
+            .build()
-                    .collect(),
+    };
-                ..Default::default()
+
-            })
+    let ice_bind_addr = SocketAddr::new(state.ice_local_addr.ip(), 0);
+    let peer: Arc<dyn PeerConnection> = Arc::new(
+        Box::pin(
+            PeerConnectionBuilder::new()
+                .with_configuration(configuration)
+                .with_handler(handler.clone())
+                .with_udp_addrs(vec![ice_bind_addr])
+                .build(),
+        )
         .await
         .map_err(|error| error.to_string())?,
     );
@@ -307,7 +278,7 @@ async fn negotiate(
         address,
         state.incoming.clone(),
     ));
-    register_peer_callbacks(&peer, &session, address);
+    *handler.session.lock().await = Some(session.clone());
 
     let offer = RTCSessionDescription::offer(offer).map_err(|error| error.to_string())?;
     peer.set_remote_description(offer)
@@ -328,14 +299,11 @@ async fn negotiate(
         .create_answer(None)
         .await
         .map_err(|error| error.to_string())?;
-    let mut gathering_complete = peer.gathering_complete_promise().await;
     peer.set_local_description(answer)
         .await
         .map_err(|error| error.to_string())?;
     trace!(%address, signaling, "Gathering NetherNet ICE candidates");
-    tokio::time::timeout(Duration::from_secs(10), gathering_complete.recv())
+    let _ = tokio::time::timeout(Duration::from_secs(2), gathering_notify.notified()).await;
-        .await
-        .map_err(|_| "Timed out gathering ICE candidates".to_string())?;
     let answer = peer
         .local_description()
         .await
@@ -351,51 +319,58 @@ async fn negotiate(
     Ok((add_server_identity(&answer, &state.identity_key)?, session))
 }
 
-fn register_peer_callbacks(
+struct NetherNetEventHandler {
-    peer: &RTCPeerConnection,
+    session: Mutex<Option<Arc<NetherNetSession>>>,
-    session: &Arc<NetherNetSession>,
     address: SocketAddr,
-) {
+    gathering_notify: Arc<tokio::sync::Notify>,
-    let session_for_channels = session.clone();
+}
-    peer.on_data_channel(Box::new(move |channel| {
+
-        let session = session_for_channels.clone();
+#[async_trait]
-        Box::pin(async move {
+impl PeerConnectionEventHandler for NetherNetEventHandler {
+    async fn on_data_channel(&self, channel: Arc<dyn DataChannel>) {
+        if let Some(session) = self.session.lock().await.as_ref() {
+            let label = channel.label().await;
+            let ordered = channel.ordered().await;
+            let negotiated = channel.negotiated().await;
+            let max_retransmits = channel.max_retransmits().await;
             trace!(
-                %address,
+                address = %self.address,
-                label = channel.label(),
+                ?label,
-                ordered = channel.ordered(),
+                ?ordered,
-                negotiated = channel.negotiated(),
+                ?negotiated,
-                max_retransmits = ?channel.max_retransmits(),
+                ?max_retransmits,
                 "Received NetherNet data channel"
             );
             if let Err(error) = session.attach_channel(channel).await {
                 warn!("Rejected NetherNet data channel: {error}");
                 session.close().await;
             }
-        })
+        }
-    }));
+    }
 
-    let session_for_state = session.clone();
+    async fn on_connection_state_change(&self, connection_state: RTCPeerConnectionState) {
-    peer.on_peer_connection_state_change(Box::new(move |connection_state| {
+        trace!(?connection_state, address = %self.address, "NetherNet peer connection state changed");
-        let session = session_for_state.clone();
-        Box::pin(async move {
-            trace!(?connection_state, %address, "NetherNet peer connection state changed");
         if matches!(
             connection_state,
             RTCPeerConnectionState::Failed
                 | RTCPeerConnectionState::Disconnected
                 | RTCPeerConnectionState::Closed
-            ) {
+        ) && let Some(session) = self.session.lock().await.as_ref()
+        {
             session.mark_closed();
         }
-        })
+    }
-    }));
 
-    peer.on_ice_connection_state_change(Box::new(move |connection_state| {
+    async fn on_ice_connection_state_change(&self, connection_state: RTCIceConnectionState) {
-        Box::pin(async move {
+        trace!(?connection_state, address = %self.address, "NetherNet ICE connection state changed");
-            trace!(?connection_state, %address, "NetherNet ICE connection state changed");
+    }
-        })
+
-    }));
+    async fn on_ice_gathering_state_change(&self, state: RTCIceGatheringState) {
+        trace!(?state, address = %self.address, "NetherNet ICE gathering state changed");
+        if state == RTCIceGatheringState::Complete {
+            self.gathering_notify.notify_waiters();
+        }
+    }
 }
 
 fn candidate_summary(sdp: &str) -> Vec<String> {
@@ -439,9 +414,9 @@ fn remove_component_two_candidates(sdp: &str) -> String {
 /// A WebRTC connection carrying complete Bedrock batch packets.
 pub struct NetherNetSession {
     #[allow(dead_code)]
-    peer: Arc<RTCPeerConnection>,
+    peer: Arc<dyn PeerConnection>,
-    reliable: RwLock<Option<Arc<RTCDataChannel>>>,
+    reliable: RwLock<Option<Arc<dyn DataChannel>>>,
-    unreliable: RwLock<Option<Arc<RTCDataChannel>>>,
+    unreliable: RwLock<Option<Arc<dyn DataChannel>>>,
     fragments: Mutex<FragmentBuffer>,
     packets: Mutex<mpsc::Receiver<Bytes>>,
     packet_sender: mpsc::Sender<Bytes>,
@@ -455,7 +430,7 @@ pub struct NetherNetSession {
 
 impl NetherNetSession {
     fn new(
-        peer: Arc<RTCPeerConnection>,
+        peer: Arc<dyn PeerConnection>,
         client_public_key: Option<PublicKey>,
         address: SocketAddr,
         incoming: mpsc::Sender<IncomingSession>,
@@ -477,23 +452,28 @@ impl NetherNetSession {
         }
     }
 
-    async fn attach_channel(self: &Arc<Self>, channel: Arc<RTCDataChannel>) -> Result<(), String> {
+    async fn attach_channel(self: &Arc<Self>, channel: Arc<dyn DataChannel>) -> Result<(), String> {
-        let has_default_parameters = channel.protocol().is_empty()
+        let label = channel.label().await.map_err(|e| e.to_string())?;
-            && !channel.negotiated()
+        let ordered = channel.ordered().await.map_err(|e| e.to_string())?;
-            && channel.max_packet_lifetime().is_none();
+        let protocol = channel.protocol().await.map_err(|e| e.to_string())?;
-        let bit = match channel.label() {
+        let negotiated = channel.negotiated().await.map_err(|e| e.to_string())?;
-            RELIABLE_CHANNEL
+        let max_packet_lifetime = channel
-                if channel.ordered()
+            .max_packet_life_time()
-                    && has_default_parameters
+            .await
-                    && channel.max_retransmits().is_none() =>
+            .map_err(|e| e.to_string())?;
-            {
+        let max_retransmits = channel.max_retransmits().await.map_err(|e| e.to_string())?;
+
+        let has_default_parameters =
+            protocol.is_empty() && !negotiated && max_packet_lifetime.is_none();
+        let bit = match label.as_str() {
+            RELIABLE_CHANNEL if ordered && has_default_parameters && max_retransmits.is_none() => {
                 *self.reliable.write().await = Some(channel.clone());
                 1
             }
             UNRELIABLE_CHANNEL
-                if !channel.ordered()
+                if !ordered
                     && has_default_parameters
-                    && channel.max_retransmits() == Some(0) =>
+                    && (max_retransmits.is_none() || max_retransmits == Some(0)) =>
             {
                 *self.unreliable.write().await = Some(channel.clone());
                 2
@@ -502,31 +482,20 @@ impl NetherNetSession {
         };
 
         let session = self.clone();
-        let channel_for_open = channel.clone();
-        channel.on_open(Box::new(move || {
-            Box::pin(async move {
-                let detached = match channel_for_open.detach().await {
-                    Ok(channel) => channel,
-                    Err(error) => {
-                        warn!(%error, address = %session.address, "Failed to detach NetherNet data channel");
-                        session.close().await;
-                        return;
-                    }
-                };
-                session.channel_opened(bit).await;
         tokio::spawn(async move {
-                    let mut buffer = vec![0; MAX_INBOUND_MESSAGE_SIZE];
+            let mut opened = false;
-                    loop {
+            while let Some(event) = channel.poll().await {
-                        match detached.read_data_channel(&mut buffer).await {
+                match event {
-                            Ok((0, _)) => break,
+                    DataChannelEvent::OnOpen => {
-                            Ok((length, _)) => {
+                        opened = true;
-                                if let Err(error) = session
+                        session.channel_opened(bit).await;
-                                    .receive_segment(
+                    }
-                                        bit,
+                    DataChannelEvent::OnMessage(msg) => {
-                                        Bytes::copy_from_slice(&buffer[..length]),
+                        if !opened {
-                                    )
+                            opened = true;
-                                    .await
+                            session.channel_opened(bit).await;
-                                {
+                        }
+                        if let Err(error) = session.receive_segment(bit, msg.data.into()).await {
                             warn!(
                                 "Invalid NetherNet message from {}: {error}",
                                 session.address
@@ -534,16 +503,17 @@ impl NetherNetSession {
                             break;
                         }
                     }
-                            Err(error) => {
+                    DataChannelEvent::OnClose => break,
-                                warn!(%error, address = %session.address, "Failed to read NetherNet data channel");
+                    DataChannelEvent::OnError => {
+                        warn!(address = %session.address, "Failed to read NetherNet data channel");
                         break;
                     }
+                    _ => {}
                 }
             }
             session.close().await;
         });
-            })
+
-        }));
         Ok(())
     }
 
@@ -625,11 +595,11 @@ impl NetherNetSession {
             return Err("Bedrock batch is too large for NetherNet".to_string());
         }
         for (index, chunk) in data.chunks(MAX_FRAGMENT_SIZE).enumerate() {
-            let mut segment = Vec::with_capacity(chunk.len() + 1);
+            let mut segment = BytesMut::with_capacity(chunk.len() + 1);
-            segment.push((segment_count - index - 1) as u8);
+            segment.put_u8((segment_count - index - 1) as u8);
             segment.extend_from_slice(chunk);
             channel
-                .send(&Bytes::from(segment))
+                .send(segment)
                 .await
                 .map_err(|error| error.to_string())?;
         }
@@ -649,11 +619,11 @@ impl NetherNetSession {
             .await
             .clone()
             .ok_or_else(|| "unreliable channel is not open".to_string())?;
-        let mut segment = Vec::with_capacity(data.len() + 1);
+        let mut segment = BytesMut::with_capacity(data.len() + 1);
-        segment.push(0);
+        segment.put_u8(0);
         segment.extend_from_slice(&data);
         channel
-            .send(&Bytes::from(segment))
+            .send(segment)
             .await
             .map_err(|error| error.to_string())?;
         Ok(())
@@ -674,12 +644,12 @@ impl NetherNetSession {
         }
     }
 
-    #[allow(clippy::unused_async)]
     pub async fn close(&self) {
         if self.closed.is_cancelled() {
             return;
         }
         self.closed.cancel();
+        let _ = self.peer.close().await;
     }
 }
 
@@ -920,11 +890,9 @@ fn unix_time() -> i64 {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::time::Duration;
     use tokio::net::UdpSocket;
-    use webrtc::{
+    use webrtc::data_channel::RTCDataChannelInit;
-        api::setting_engine::SctpMaxMessageSize,
-        data_channel::data_channel_init::RTCDataChannelInit,
-    };
 
     #[test]
     fn fragments_round_trip() {
@@ -1056,23 +1024,33 @@ mod tests {
             .unwrap()
     }
 
-    fn test_client_api() -> API {
+    struct ClientHandler {
-        let mut media_engine = MediaEngine::default();
+        notify: Arc<tokio::sync::Notify>,
-        media_engine.register_default_codecs().unwrap();
+    }
-        let mut setting_engine = SettingEngine::default();
+    #[async_trait]
-        setting_engine.set_sctp_max_message_size_can_send(SctpMaxMessageSize::Unbounded);
+    impl PeerConnectionEventHandler for ClientHandler {
-        APIBuilder::new()
+        async fn on_ice_gathering_state_change(&self, state: RTCIceGatheringState) {
-            .with_media_engine(media_engine)
+            if state == RTCIceGatheringState::Complete {
-            .with_setting_engine(setting_engine)
+                self.notify.notify_waiters();
-            .build()
+            }
+        }
     }
 
     #[tokio::test]
+    #[allow(clippy::too_many_lines)]
     async fn negotiates_channels_and_receives_a_packet() {
         let _ = tracing_subscriber::fmt().with_test_writer().try_init();
-        let client = Arc::new(
+        let client_notify = Arc::new(tokio::sync::Notify::new());
-            test_client_api()
+        let client: Arc<dyn PeerConnection> = Arc::new(
-                .new_peer_connection(RTCConfiguration::default())
+            Box::pin(
+                PeerConnectionBuilder::new()
+                    .with_configuration(RTCConfigurationBuilder::default().build())
+                    .with_handler(Arc::new(ClientHandler {
+                        notify: client_notify.clone(),
+                    }))
+                    .with_udp_addrs(vec!["127.0.0.1:0"])
+                    .build(),
+            )
             .await
             .unwrap(),
         );
@@ -1080,7 +1058,7 @@ mod tests {
             .create_data_channel(
                 RELIABLE_CHANNEL,
                 Some(RTCDataChannelInit {
-                    ordered: Some(true),
+                    ordered: true,
                     ..Default::default()
                 }),
             )
@@ -1090,7 +1068,7 @@ mod tests {
             .create_data_channel(
                 UNRELIABLE_CHANNEL,
                 Some(RTCDataChannelInit {
-                    ordered: Some(false),
+                    ordered: false,
                     max_retransmits: Some(0),
                     ..Default::default()
                 }),
@@ -1098,66 +1076,64 @@ mod tests {
             .await
             .unwrap();
         let (unreliable_sender, mut unreliable_receiver) = mpsc::channel(1);
-        unreliable.on_message(Box::new(move |message| {
+        let unreliable_poller = unreliable.clone();
-            let sender = unreliable_sender.clone();
+        tokio::spawn(async move {
-            Box::pin(async move {
+            while let Some(event) = unreliable_poller.poll().await {
-                let _ = sender.send(message.data).await;
+                if let DataChannelEvent::OnMessage(msg) = event {
-            })
+                    let _ = unreliable_sender.send(msg.data.into()).await;
-        }));
+                }
+            }
+        });
         let offer = client.create_offer(None).await.unwrap();
-        let mut gathering_complete = client.gathering_complete_promise().await;
         client.set_local_description(offer).await.unwrap();
-        gathering_complete.recv().await;
+        let _ = tokio::time::timeout(Duration::from_secs(2), client_notify.notified()).await;
         let offer = client.local_description().await.unwrap();
         let client_key = SigningKey::from_slice(&[8; 48]).unwrap();
         let offer = add_server_identity(&offer.sdp, &client_key).unwrap();
         let (incoming, mut receiver) = mpsc::channel(1);
         let server_key = Arc::new(SigningKey::from_slice(&[9; 48]).unwrap());
         let ice_socket = UdpSocket::bind("0.0.0.0:0").await.unwrap();
-        let ice_port = ice_socket.local_addr().unwrap().port();
+        let ice_local_addr = ice_socket.local_addr().unwrap();
         let state = EndpointState {
             incoming,
-            api: Arc::new(build_api(ice_socket, None).unwrap()),
             identity_key: server_key.clone(),
             require_client_identity: true,
             oidc_verifier: None,
             stun_servers: Arc::from([]),
+            ice_local_addr,
         };
-        let (answer, server_session) =
+        let (answer, _server_session) =
             negotiate(&state, "127.0.0.1:19132".parse().unwrap(), &offer, None)
                 .await
                 .unwrap();
         let (answer, public_key) = verify_and_strip_identity(&answer, None).unwrap();
         assert_eq!(public_key, PublicKey::from(server_key.verifying_key()));
-        assert!(answer.contains(&format!(" {ice_port} typ host")));
-        let answer = answer.replace(
-            "a=sctp-port:5000\r\n",
-            "a=sctp-port:5000\r\na=max-message-size:262144\r\n",
-        );
         client
             .set_remote_description(RTCSessionDescription::answer(answer).unwrap())
             .await
             .unwrap();
+        let reliable_poller = reliable.clone();
+        tokio::spawn(async move { while reliable_poller.poll().await.is_some() {} });
         let Ok(Some((session, _))) =
             tokio::time::timeout(Duration::from_secs(5), receiver.recv()).await
         else {
-            panic!(
+            panic!("connection did not open");
-                "connection did not open; client={:?}, server={:?}",
-                client.connection_state(),
-                server_session.peer.connection_state(),
-            );
         };
         reliable
-            .send(&Bytes::from_static(b"\0hello"))
+            .send(BytesMut::from(&b"\0hello"[..]))
             .await
             .unwrap();
         let packet = receive_packet(&session).await;
         assert_eq!(packet, b"hello".as_slice());
         let large_packet = vec![42; 100_000];
-        let mut segment = Vec::with_capacity(large_packet.len() + 1);
+        let chunks = large_packet.chunks(10_000).collect::<Vec<_>>();
-        segment.push(0);
+        let chunk_count = chunks.len();
-        segment.extend_from_slice(&large_packet);
+        for (index, chunk) in chunks.into_iter().enumerate() {
-        reliable.send(&Bytes::from(segment)).await.unwrap();
+            let mut segment = BytesMut::with_capacity(chunk.len() + 1);
+            segment.put_u8((chunk_count - index - 1) as u8);
+            segment.extend_from_slice(chunk);
+            reliable.send(segment).await.unwrap();
+        }
         let packet = receive_packet(&session).await;
         assert_eq!(packet, large_packet);
         session

crates/pumpkin/src/net/bedrock/nethernet/discovery.rs

@@ -17,7 +17,7 @@ use tokio::{
     sync::{Mutex, mpsc},
 };
 use tracing::{debug, trace, warn};
-use webrtc::ice_transport::ice_candidate::RTCIceCandidateInit;
+use webrtc::peer_connection::RTCIceCandidateInit;
 
 use super::{NetherNetListener, negotiate};
 use crate::server::Server;
@@ -169,8 +169,14 @@ impl NetherNetDiscovery {
                 let offer = data.to_owned();
                 let network_id = self.network_id;
                 tokio::spawn(async move {
-                    let signal =
+                    let signal = match Box::pin(negotiate(
-                        match negotiate(&state, address, &offer, Some(candidate_receiver)).await {
+                        &state,
+                        address,
+                        &offer,
+                        Some(candidate_receiver),
+                    ))
+                    .await
+                    {
                         Ok((answer, _session)) => {
                             format!("CONNECTRESPONSE {connection_id} {answer}")
                         }

crates/pumpkin/src/net/bedrock/status.rs

@@ -1,11 +1,10 @@
 use std::{
-    future::Future,
+    io::{Cursor, Error},
-    io::{Cursor, Error, ErrorKind},
     net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
-    pin::Pin,
     sync::Arc,
 };
 
+use crate::server::Server;
 use bytes::Bytes;
 use pumpkin_protocol::{
     BClientPacket,
@@ -21,14 +20,7 @@ use tokio::{
     net::UdpSocket,
     sync::{Mutex, mpsc},
 };
-use tracing::{trace, warn};
+use tracing::trace;
-use webrtc::util::{Conn, Error as WebRtcError};
-
-use crate::server::Server;
-
-// `webrtc::util::Conn` uses `async-trait`. Spell out its object-safe future ABI so
-// Pumpkin does not need a direct dependency on the proc macro.
-type ConnFuture<'a, T> = Pin<Box<dyn Future<Output = Result<T, WebRtcError>> + Send + 'a>>;
 
 pub struct StatusResponder {
     ipv4: Arc<UdpSocket>,
@@ -147,119 +139,18 @@ impl IceSocket {
     pub fn local_addr(&self) -> Result<SocketAddr, Error> {
         self.socket.local_addr()
     }
-}
 
-impl Conn for IceSocket {
+    pub async fn recv_from(&self, buffer: &mut [u8]) -> Result<(usize, SocketAddr), Error> {
-    fn connect<'a, 'async_trait>(&'a self, _address: SocketAddr) -> ConnFuture<'async_trait, ()>
+        let (packet, address) = self.packets.lock().await.recv().await.ok_or_else(|| {
-    where
+            Error::new(std::io::ErrorKind::BrokenPipe, "Bedrock UDP socket closed")
-        'a: 'async_trait,
-        Self: 'async_trait,
-    {
-        Box::pin(async {
-            Err(Error::new(
-                ErrorKind::Unsupported,
-                "the shared Bedrock UDP socket cannot be connected",
-            )
-            .into())
-        })
-    }
-
-    fn recv<'a, 'b, 'async_trait>(&'a self, buffer: &'b mut [u8]) -> ConnFuture<'async_trait, usize>
-    where
-        'a: 'async_trait,
-        'b: 'async_trait,
-        Self: 'async_trait,
-    {
-        Box::pin(async move { self.recv_from(buffer).await.map(|(length, _)| length) })
-    }
-
-    fn recv_from<'a, 'b, 'async_trait>(
-        &'a self,
-        buffer: &'b mut [u8],
-    ) -> ConnFuture<'async_trait, (usize, SocketAddr)>
-    where
-        'a: 'async_trait,
-        'b: 'async_trait,
-        Self: 'async_trait,
-    {
-        Box::pin(async move {
-            let (packet, address) =
-                self.packets.lock().await.recv().await.ok_or_else(|| {
-                    Error::new(ErrorKind::BrokenPipe, "Bedrock UDP socket closed")
         })?;
         let length = buffer.len().min(packet.len());
         buffer[..length].copy_from_slice(&packet[..length]);
         Ok((length, address))
-        })
     }
 
-    fn send<'a, 'b, 'async_trait>(&'a self, _buffer: &'b [u8]) -> ConnFuture<'async_trait, usize>
+    pub async fn send_to(&self, buffer: &[u8], target: SocketAddr) -> Result<usize, Error> {
-    where
+        self.socket.send_to(buffer, target).await
-        'a: 'async_trait,
-        'b: 'async_trait,
-        Self: 'async_trait,
-    {
-        Box::pin(async {
-            Err(Error::new(
-                ErrorKind::NotConnected,
-                "the shared Bedrock UDP socket has no default peer",
-            )
-            .into())
-        })
-    }
-
-    fn send_to<'a, 'b, 'async_trait>(
-        &'a self,
-        buffer: &'b [u8],
-        target: SocketAddr,
-    ) -> ConnFuture<'async_trait, usize>
-    where
-        'a: 'async_trait,
-        'b: 'async_trait,
-        Self: 'async_trait,
-    {
-        Box::pin(async move {
-            match self.socket.send_to(buffer, target).await {
-                Ok(length) => {
-                    trace!(
-                        %target,
-                        length,
-                        kind = ice_packet_kind(buffer),
-                        "Sent Bedrock ICE datagram"
-                    );
-                    Ok(length)
-                }
-                Err(error) => {
-                    warn!(
-                        %target,
-                        kind = ice_packet_kind(buffer),
-                        %error,
-                        "Failed to send Bedrock ICE datagram"
-                    );
-                    Err(error.into())
-                }
-            }
-        })
-    }
-
-    fn local_addr(&self) -> Result<SocketAddr, WebRtcError> {
-        Ok(self.socket.local_addr()?)
-    }
-
-    fn remote_addr(&self) -> Option<SocketAddr> {
-        None
-    }
-
-    fn close<'a, 'async_trait>(&'a self) -> ConnFuture<'async_trait, ()>
-    where
-        'a: 'async_trait,
-        Self: 'async_trait,
-    {
-        Box::pin(async { Ok(()) })
-    }
-
-    fn as_any(&self) -> &(dyn std::any::Any + Send + Sync) {
-        self
     }
 }
 
@@ -357,11 +248,11 @@ mod tests {
             .unwrap();
 
         let mut request = [0; 16];
-        let (length, address) = Conn::recv_from(&ice, &mut request).await.unwrap();
+        let (length, address) = ice.recv_from(&mut request).await.unwrap();
         assert_eq!(&request[..length], b"request");
         assert_eq!(address, client.local_addr().unwrap());
 
-        Conn::send_to(&ice, b"response", client.local_addr().unwrap())
+        ice.send_to(b"response", client.local_addr().unwrap())
             .await
             .unwrap();
         let mut response = [0; 16];