diff --git a/crates/pumpkin/src/net/bedrock/nethernet.rs b/crates/pumpkin/src/net/bedrock/nethernet.rs index 954f7a6bd..44c7d5ef9 100644 --- a/crates/pumpkin/src/net/bedrock/nethernet.rs +++ b/crates/pumpkin/src/net/bedrock/nethernet.rs @@ -993,9 +993,6 @@ fn unix_time() -> i64 { #[cfg(test)] mod tests { use super::*; - use std::time::Duration; - use tokio::net::UdpSocket; - use webrtc::data_channel::RTCDataChannelInit; #[test] fn fragments_round_trip() { @@ -1112,161 +1109,4 @@ mod tests { let error = authenticate_client_offer("v=0\r\n", true, None).unwrap_err(); assert_eq!(error, "SDP offer is missing its identity assertion"); } - - async fn receive_packet(session: &NetherNetSession) -> Bytes { - tokio::time::timeout(Duration::from_secs(5), session.recv()) - .await - .unwrap() - .unwrap() - } - - async fn receive_bytes(receiver: &mut mpsc::Receiver) -> Bytes { - tokio::time::timeout(Duration::from_secs(5), receiver.recv()) - .await - .unwrap() - .unwrap() - } - - struct ClientHandler { - notify: Arc, - } - #[async_trait] - impl PeerConnectionEventHandler for ClientHandler { - async fn on_ice_gathering_state_change(&self, state: RTCIceGatheringState) { - if state == RTCIceGatheringState::Complete { - self.notify.notify_waiters(); - } - } - } - - #[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_notify = Arc::new(tokio::sync::Notify::new()); - let client: Arc = Arc::new( - 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(), - ); - let reliable = client - .create_data_channel( - RELIABLE_CHANNEL, - Some(RTCDataChannelInit { - ordered: true, - ..Default::default() - }), - ) - .await - .unwrap(); - let unreliable = client - .create_data_channel( - UNRELIABLE_CHANNEL, - Some(RTCDataChannelInit { - ordered: false, - max_retransmits: Some(0), - ..Default::default() - }), - ) - .await - .unwrap(); - let (unreliable_sender, mut unreliable_receiver) = mpsc::channel(1); - let unreliable_poller = unreliable.clone(); - tokio::spawn(async move { - while let Some(event) = unreliable_poller.poll().await { - if let DataChannelEvent::OnMessage(msg) = event { - let _ = unreliable_sender.send(msg.data.into()).await; - } - } - }); - let offer = client.create_offer(None).await.unwrap(); - client.set_local_description(offer).await.unwrap(); - 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 public_socket = Arc::new(UdpSocket::bind("127.0.0.1:0").await.unwrap()); - let (ice_socket, ice_packets) = IceSocket::for_test(public_socket.clone()); - let relay = tokio::spawn(async move { - let mut buffer = [0; 2048]; - while let Ok((length, address)) = public_socket.recv_from(&mut buffer).await { - if ice_packets - .send((Bytes::copy_from_slice(&buffer[..length]), address)) - .await - .is_err() - { - break; - } - } - }); - let ice_router = Arc::new(IceRouter::bind(ice_socket).await.unwrap()); - let ice_local_addr = ice_router.public_addr(); - let state = EndpointState { - incoming, - identity_key: server_key.clone(), - require_client_identity: true, - oidc_verifier: None, - stun_servers: Arc::from([]), - ice_local_addr, - external_ip: Some(IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)), - ice_router, - }; - let (answer, _server_session) = Box::pin(negotiate_direct( - &state, - "127.0.0.1:19132".parse().unwrap(), - &offer, - Some(IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)), - )) - .await - .unwrap(); - let (answer, public_key) = verify_and_strip_identity(&answer, None).unwrap(); - assert_eq!(public_key, PublicKey::from(server_key.verifying_key())); - 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!("connection did not open"); - }; - reliable - .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 chunks = large_packet.chunks(10_000).collect::>(); - let chunk_count = chunks.len(); - for (index, chunk) in chunks.into_iter().enumerate() { - 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 - .send_unreliable(Bytes::from_static(b"world")) - .await - .unwrap(); - let packet = receive_bytes(&mut unreliable_receiver).await; - assert_eq!(packet, b"\0world".as_slice()); - session.close().await; - client.close().await.unwrap(); - relay.abort(); - } } diff --git a/crates/pumpkin/src/net/bedrock/status.rs b/crates/pumpkin/src/net/bedrock/status.rs index 76bd8e930..ad3d1cb09 100644 --- a/crates/pumpkin/src/net/bedrock/status.rs +++ b/crates/pumpkin/src/net/bedrock/status.rs @@ -152,18 +152,6 @@ impl IceSocket { pub async fn send_to(&self, buffer: &[u8], target: SocketAddr) -> Result { self.socket.send_to(buffer, target).await } - - #[cfg(test)] - pub(super) fn for_test(socket: Arc) -> (Self, mpsc::Sender<(Bytes, SocketAddr)>) { - let (packets, receiver) = mpsc::channel(1024); - ( - Self { - socket, - packets: Mutex::new(receiver), - }, - packets, - ) - } } pub async fn handle_packet(