Skip to main content

RaknetSocket

rust_raknet

Struct RaknetSocket 

Source 
pub struct RaknetSocket { /* private fields */ }
Expand description

A RakNet connection backed by a UDP socket.

Implementations§

Source§

impl RaknetSocket

Source

pub fn set_idle_maintenance(&self, enabled: bool)

Pause maintenance after 500 ms of quiet traffic and empty queues.

Disabled by default to preserve the periodic schedule under load. This can save CPU with many idle peers, but changes maintenance timing. Sending or receiving work wakes a paused connection. Disabling this option immediately resumes periodic maintenance.

Source

pub fn mtu(&self) -> u16

Return the negotiated nominal RakNet MTU in bytes.

Source

pub async fn from( addr: &SocketAddr, s: &Arc<UdpSocket>, receiver: Receiver<Vec<u8>>, mtu: u16, collecter: Arc<Mutex<Sender<SocketAddr>>>, raknet_version: u8, ) -> Self

Create a RakNet socket for an established UDP connection.

This constructor is used internally by [RaknetListener].

Source

pub async fn connect(addr: &SocketAddr) -> Result<Self>

Connect to a RakNet server.

§Example
ⓘ
let socket = RaknetSocket::connect("127.0.0.1:19132".parse().unwrap()).await.unwrap();
socket.send(&[0xfe], Reliability::ReliableOrdered).await.unwrap();
let buf = socket.recv().await.unwrap();
if buf[0] == 0xfe{
   //do something
}
Source

pub async fn connect_with_version( addr: &SocketAddr, raknet_version: u8, ) -> Result<Self>

Examples found in repository?
examples/concurrency_benchmark.rs (line 280)
270    async fn connect(address: std::net::SocketAddr, tcp: bool) -> Result<Self, BenchError> {
271        if tcp {
272            let stream = TcpStream::connect(address).await?;
273            stream.set_nodelay(true)?;
274            Ok(Self::Tcp {
275                stream,
276                record: Vec::new(),
277            })
278        } else {
279            Ok(Self::RakNet(
280                RaknetSocket::connect_with_version(&address, 11).await?,
281            ))
282        }
283    }
More examples
Hide additional examples
examples/bedrock_probe.rs (line 12)
4async fn main() {
5    let args: Vec<_> = env::args().collect();
6    let address = args[1].parse().unwrap();
7    let version: u8 = args[2].parse().unwrap();
8    rust_raknet::enable_raknet_log(3);
9    println!("TARGET {address} RAKNET {version}");
10    let socket = match tokio::time::timeout(
11        Duration::from_secs(10),
12        RaknetSocket::connect_with_version(&address, version),
13    )
14    .await
15    {
16        Ok(Ok(socket)) => {
17            println!("HANDSHAKE OK");
18            socket
19        }
20        Ok(Err(error)) => {
21            println!("HANDSHAKE ERROR {error}");
22            std::process::exit(1);
23        }
24        Err(_) => {
25            println!("HANDSHAKE TIMEOUT");
26            std::process::exit(2);
27        }
28    };
29    if let Some(protocol) = args.get(3) {
30        let protocol: i32 = protocol.parse().unwrap();
31        let mut request = vec![0xfe, 6, 0xc1, 1];
32        request.extend_from_slice(&protocol.to_be_bytes());
33        socket
34            .send(&request, Reliability::ReliableOrdered)
35            .await
36            .unwrap();
37        println!("SENT RequestNetworkSettings protocol={protocol}");
38        match tokio::time::timeout(Duration::from_secs(5), socket.recv()).await {
39            Ok(Ok(data)) => {
40                println!(
41                    "RECEIVED {} bytes: {:02x?}",
42                    data.len(),
43                    &data[..data.len().min(80)]
44                );
45                if data.len() >= 8 && data[0] == 0xfe && data[2..4] == [0x8f, 1] {
46                    println!(
47                        "NETWORK_SETTINGS OK threshold={} algorithm={}",
48                        u16::from_le_bytes([data[4], data[5]]),
49                        u16::from_le_bytes([data[6], data[7]])
50                    );
51                } else {
52                    println!("UNEXPECTED APPLICATION RESPONSE");
53                    std::process::exit(3);
54                }
55            }
56            other => {
57                println!("APPLICATION RECEIVE ERROR {other:?}");
58                std::process::exit(4);
59            }
60        }
61    }
62    let _ = tokio::time::timeout(Duration::from_secs(1), socket.close()).await;
63}
examples/interop.rs (line 16)
6async fn main() -> Result<(), Box<dyn Error>> {
7    if std::env::var_os("RAKNET_DEBUG").is_some() {
8        rust_raknet::enable_raknet_log(7);
9    }
10    let address = std::env::args()
11        .nth(1)
12        .ok_or("missing peer address")?
13        .parse()?;
14    let batch = std::env::args().nth(2).as_deref() == Some("--batch");
15    tokio::time::timeout(Duration::from_secs(60), async {
16        let socket = RaknetSocket::connect_with_version(&address, 11).await?;
17        let mut count = 0;
18        for size in [64, 800, 4096, 10000] {
19            if batch && size <= 800 {
20                for first in (0..100).step_by(16) {
21                    let messages: Vec<_> = (first..(first + 16).min(100))
22                        .map(|index| {
23                            let mut message: Vec<_> =
24                                (0..size).map(|i| (i + index) as u8).collect();
25                            message[0] = 0xfe;
26                            rust_raknet::Bytes::from(message)
27                        })
28                        .collect();
29                    socket
30                        .send_bytes_batch(&messages, Reliability::ReliableOrdered)
31                        .await?;
32                    for expected in messages {
33                        assert_eq!(
34                            socket.recv().await?.as_slice(),
35                            expected.as_ref(),
36                            "message reordered, duplicated or corrupted"
37                        );
38                        count += 1;
39                    }
40                }
41            } else {
42                for index in 0..100 {
43                    let mut payload: Vec<_> = (0..size).map(|i| (i + index) as u8).collect();
44                    payload[0] = 0xfe;
45                    socket.send(&payload, Reliability::ReliableOrdered).await?;
46                    assert_eq!(
47                        socket.recv().await?,
48                        payload,
49                        "message reordered, duplicated or corrupted"
50                    );
51                    count += 1;
52                }
53            }
54        }
55        socket.close().await?;
56        println!("PASS: {count} unique ordered messages, 64/800/4096/10000 bytes");
57        Ok::<_, Box<dyn Error>>(())
58    })
59    .await?
60}
examples/concurrency.rs (line 30)
11async fn main() -> Result<(), Box<dyn Error>> {
12    let args: Vec<_> = std::env::args().collect();
13    let address = args
14        .get(1)
15        .ok_or("usage: concurrency ADDRESS [CONNECTIONS] [MESSAGES]")?
16        .parse()?;
17    let connections: usize = args.get(2).map_or(Ok(512), |n| n.parse())?;
18    let messages: usize = args.get(3).map_or(Ok(100), |n| n.parse())?;
19    if connections == 0 || messages == 0 {
20        return Err("counts must be positive".into());
21    }
22    tokio::time::timeout(Duration::from_secs(120), async {
23        let started = Instant::now();
24        let dialing = Arc::new(Semaphore::new(8));
25        let mut tasks = JoinSet::new();
26        for index in 0..connections {
27            let dialing = dialing.clone();
28            tasks.spawn(async move {
29                let _permit = dialing.acquire().await.unwrap();
30                let socket = RaknetSocket::connect_with_version(&address, 11).await?;
31                Ok::<_, rust_raknet::error::RaknetError>((index, socket))
32            });
33        }
34        let mut sockets = Vec::with_capacity(connections);
35        while let Some(result) = tasks.join_next().await { sockets.push(result??); }
36        let handshake_seconds = started.elapsed().as_secs_f64();
37        let transfer = Instant::now();
38        let mut exchanges = JoinSet::new();
39        for (index, socket) in sockets {
40            exchanges.spawn(async move {
41                fn payload(connection: usize, message: usize) -> Vec<u8> {
42                    let size = [64, 800, 4096][message % 3];
43                    let mut bytes = vec![0xfe; size];
44                    bytes[1..9].copy_from_slice(&(connection as u64).to_le_bytes());
45                    bytes[9..17].copy_from_slice(&(message as u64).to_le_bytes());
46                    bytes
47                }
48                let sender = async {
49                    for message in 0..messages {
50                        socket.send(&payload(index, message), Reliability::ReliableOrdered).await?;
51                    }
52                    socket.flush().await
53                };
54                let receiver = async {
55                    // Exercise a slow application without blocking protocol ACKs.
56                    tokio::time::sleep(Duration::from_millis(30)).await;
57                    for message in 0..messages {
58                        let actual = socket.recv().await?;
59                        if actual != payload(index, message) {
60                            return Err(rust_raknet::error::RaknetError::PacketParseError);
61                        }
62                    }
63                    Ok(())
64                };
65                tokio::try_join!(sender, receiver)?;
66                socket.close().await
67            });
68        }
69        while let Some(result) = exchanges.join_next().await { result??; }
70        println!("PASS: {connections} simultaneous connections, {} unique ordered echoes; handshake={handshake_seconds:.3}s, transfer={:.3}s", connections * messages, transfer.elapsed().as_secs_f64());
71        Ok::<_, Box<dyn Error>>(())
72    }).await?
73}
Source

pub async fn connect_with_version_and_mtu( addr: &SocketAddr, raknet_version: u8, requested_mtu: u16, ) -> Result<Self>

Connect with an explicit RakNet version and nominal MTU in bytes.

The requested MTU must be between 61 and 1,492 bytes. The server may negotiate a smaller value. Choose a size the network path can carry; the ordinary connection APIs retain their 1,400-byte default.

Source

pub async fn close(&self) -> Result<()>

Close the RakNet connection. The connection also closes when the socket is dropped. This method is idempotent.

§Example
ⓘ
let (latency, motd) = socket::RaknetSocket::ping("127.0.0.1:19132".parse().unwrap()).await.unwrap();
assert!((0..10).contains(&latency));
Examples found in repository?
examples/bedrock_probe.rs (line 62)
4async fn main() {
5    let args: Vec<_> = env::args().collect();
6    let address = args[1].parse().unwrap();
7    let version: u8 = args[2].parse().unwrap();
8    rust_raknet::enable_raknet_log(3);
9    println!("TARGET {address} RAKNET {version}");
10    let socket = match tokio::time::timeout(
11        Duration::from_secs(10),
12        RaknetSocket::connect_with_version(&address, version),
13    )
14    .await
15    {
16        Ok(Ok(socket)) => {
17            println!("HANDSHAKE OK");
18            socket
19        }
20        Ok(Err(error)) => {
21            println!("HANDSHAKE ERROR {error}");
22            std::process::exit(1);
23        }
24        Err(_) => {
25            println!("HANDSHAKE TIMEOUT");
26            std::process::exit(2);
27        }
28    };
29    if let Some(protocol) = args.get(3) {
30        let protocol: i32 = protocol.parse().unwrap();
31        let mut request = vec![0xfe, 6, 0xc1, 1];
32        request.extend_from_slice(&protocol.to_be_bytes());
33        socket
34            .send(&request, Reliability::ReliableOrdered)
35            .await
36            .unwrap();
37        println!("SENT RequestNetworkSettings protocol={protocol}");
38        match tokio::time::timeout(Duration::from_secs(5), socket.recv()).await {
39            Ok(Ok(data)) => {
40                println!(
41                    "RECEIVED {} bytes: {:02x?}",
42                    data.len(),
43                    &data[..data.len().min(80)]
44                );
45                if data.len() >= 8 && data[0] == 0xfe && data[2..4] == [0x8f, 1] {
46                    println!(
47                        "NETWORK_SETTINGS OK threshold={} algorithm={}",
48                        u16::from_le_bytes([data[4], data[5]]),
49                        u16::from_le_bytes([data[6], data[7]])
50                    );
51                } else {
52                    println!("UNEXPECTED APPLICATION RESPONSE");
53                    std::process::exit(3);
54                }
55            }
56            other => {
57                println!("APPLICATION RECEIVE ERROR {other:?}");
58                std::process::exit(4);
59            }
60        }
61    }
62    let _ = tokio::time::timeout(Duration::from_secs(1), socket.close()).await;
63}
More examples
Hide additional examples
examples/interop.rs (line 55)
6async fn main() -> Result<(), Box<dyn Error>> {
7    if std::env::var_os("RAKNET_DEBUG").is_some() {
8        rust_raknet::enable_raknet_log(7);
9    }
10    let address = std::env::args()
11        .nth(1)
12        .ok_or("missing peer address")?
13        .parse()?;
14    let batch = std::env::args().nth(2).as_deref() == Some("--batch");
15    tokio::time::timeout(Duration::from_secs(60), async {
16        let socket = RaknetSocket::connect_with_version(&address, 11).await?;
17        let mut count = 0;
18        for size in [64, 800, 4096, 10000] {
19            if batch && size <= 800 {
20                for first in (0..100).step_by(16) {
21                    let messages: Vec<_> = (first..(first + 16).min(100))
22                        .map(|index| {
23                            let mut message: Vec<_> =
24                                (0..size).map(|i| (i + index) as u8).collect();
25                            message[0] = 0xfe;
26                            rust_raknet::Bytes::from(message)
27                        })
28                        .collect();
29                    socket
30                        .send_bytes_batch(&messages, Reliability::ReliableOrdered)
31                        .await?;
32                    for expected in messages {
33                        assert_eq!(
34                            socket.recv().await?.as_slice(),
35                            expected.as_ref(),
36                            "message reordered, duplicated or corrupted"
37                        );
38                        count += 1;
39                    }
40                }
41            } else {
42                for index in 0..100 {
43                    let mut payload: Vec<_> = (0..size).map(|i| (i + index) as u8).collect();
44                    payload[0] = 0xfe;
45                    socket.send(&payload, Reliability::ReliableOrdered).await?;
46                    assert_eq!(
47                        socket.recv().await?,
48                        payload,
49                        "message reordered, duplicated or corrupted"
50                    );
51                    count += 1;
52                }
53            }
54        }
55        socket.close().await?;
56        println!("PASS: {count} unique ordered messages, 64/800/4096/10000 bytes");
57        Ok::<_, Box<dyn Error>>(())
58    })
59    .await?
60}
examples/concurrency.rs (line 66)
11async fn main() -> Result<(), Box<dyn Error>> {
12    let args: Vec<_> = std::env::args().collect();
13    let address = args
14        .get(1)
15        .ok_or("usage: concurrency ADDRESS [CONNECTIONS] [MESSAGES]")?
16        .parse()?;
17    let connections: usize = args.get(2).map_or(Ok(512), |n| n.parse())?;
18    let messages: usize = args.get(3).map_or(Ok(100), |n| n.parse())?;
19    if connections == 0 || messages == 0 {
20        return Err("counts must be positive".into());
21    }
22    tokio::time::timeout(Duration::from_secs(120), async {
23        let started = Instant::now();
24        let dialing = Arc::new(Semaphore::new(8));
25        let mut tasks = JoinSet::new();
26        for index in 0..connections {
27            let dialing = dialing.clone();
28            tasks.spawn(async move {
29                let _permit = dialing.acquire().await.unwrap();
30                let socket = RaknetSocket::connect_with_version(&address, 11).await?;
31                Ok::<_, rust_raknet::error::RaknetError>((index, socket))
32            });
33        }
34        let mut sockets = Vec::with_capacity(connections);
35        while let Some(result) = tasks.join_next().await { sockets.push(result??); }
36        let handshake_seconds = started.elapsed().as_secs_f64();
37        let transfer = Instant::now();
38        let mut exchanges = JoinSet::new();
39        for (index, socket) in sockets {
40            exchanges.spawn(async move {
41                fn payload(connection: usize, message: usize) -> Vec<u8> {
42                    let size = [64, 800, 4096][message % 3];
43                    let mut bytes = vec![0xfe; size];
44                    bytes[1..9].copy_from_slice(&(connection as u64).to_le_bytes());
45                    bytes[9..17].copy_from_slice(&(message as u64).to_le_bytes());
46                    bytes
47                }
48                let sender = async {
49                    for message in 0..messages {
50                        socket.send(&payload(index, message), Reliability::ReliableOrdered).await?;
51                    }
52                    socket.flush().await
53                };
54                let receiver = async {
55                    // Exercise a slow application without blocking protocol ACKs.
56                    tokio::time::sleep(Duration::from_millis(30)).await;
57                    for message in 0..messages {
58                        let actual = socket.recv().await?;
59                        if actual != payload(index, message) {
60                            return Err(rust_raknet::error::RaknetError::PacketParseError);
61                        }
62                    }
63                    Ok(())
64                };
65                tokio::try_join!(sender, receiver)?;
66                socket.close().await
67            });
68        }
69        while let Some(result) = exchanges.join_next().await { result??; }
70        println!("PASS: {connections} simultaneous connections, {} unique ordered echoes; handshake={handshake_seconds:.3}s, transfer={:.3}s", connections * messages, transfer.elapsed().as_secs_f64());
71        Ok::<_, Box<dyn Error>>(())
72    }).await?
73}
Source

pub async fn ping(addr: &SocketAddr) -> Result<(i64, String)>

Ping a RakNet server and return the round-trip time and MOTD.

§Example
ⓘ
let (latency, motd) = socket::RaknetSocket::ping("127.0.0.1:19132".parse().unwrap()).await.unwrap();
assert!((0..10).contains(&latency));
Source

pub async fn send(&self, buf: &[u8], reliability: Reliability) -> Result<()>

Send a Bedrock game packet.

The packet must begin with 0xfe. ReliableOrdered messages are fragmented when necessary; other modes must fit the negotiated MTU. Messages exceeding 65,536 fragments or the 64 MiB send budget (including fragment overhead) return RaknetError::PacketSizeExceedMTU. A full send queue applies asynchronous backpressure until capacity is available or the connection closes.

§Example
ⓘ
let socket = RaknetSocket::connect("127.0.0.1:19132".parse().unwrap()).await.unwrap();
socket.send(&[0xfe], Reliability::ReliableOrdered).await.unwrap();
Examples found in repository?
examples/concurrency_benchmark.rs (line 287)
285    async fn send(&mut self, payload: &[u8]) -> Result<(), BenchError> {
286        match self {
287            Self::RakNet(socket) => socket.send(payload, Reliability::ReliableOrdered).await?,
288            Self::Tcp { stream, record, .. } => {
289                let length = u32::try_from(payload.len())?;
290                record.clear();
291                record.extend_from_slice(&length.to_le_bytes());
292                record.extend_from_slice(payload);
293                stream.write_all(record).await?;
294            }
295        }
296        Ok(())
297    }
More examples
Hide additional examples
examples/sharded_echo.rs (line 48)
4async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
5    use rust_raknet::{RaknetListener, Reliability};
6    let address = std::env::args().nth(1).ok_or("ADDRESS required")?.parse()?;
7    let count = std::env::args()
8        .nth(2)
9        .unwrap_or_else(|| "4".into())
10        .parse()?;
11    let mut listener = RaknetListener::bind_with_socket_shards(&address, count).await?;
12    let mut batch_messages = false;
13    for option in std::env::args().skip(3) {
14        listener = match option.as_str() {
15            "--receive-batching" => listener.with_receive_batching(true),
16            "--idle-maintenance" => listener.with_idle_maintenance(true),
17            "--batch-messages" => {
18                batch_messages = true;
19                listener
20            }
21            _ => return Err(format!("unknown option: {option}").into()),
22        };
23    }
24    listener.listen().await;
25    println!("Sharded echo listening on {}", listener.local_addr()?);
26    loop {
27        let client = listener.accept().await?;
28        if batch_messages {
29            tokio::spawn(async move {
30                let mut messages = Vec::with_capacity(16);
31                while client.recv_bytes_batch(&mut messages, 16).await.is_ok() {
32                    if client
33                        .send_bytes_batch(&messages, Reliability::ReliableOrdered)
34                        .await
35                        .is_err()
36                    {
37                        break;
38                    }
39                    if messages.len() == 16 {
40                        tokio::task::yield_now().await;
41                    }
42                }
43            });
44        } else {
45            tokio::spawn(async move {
46                while let Ok(payload) = client.recv().await {
47                    if client
48                        .send(&payload, Reliability::ReliableOrdered)
49                        .await
50                        .is_err()
51                    {
52                        break;
53                    }
54                }
55            });
56        }
57    }
58}
examples/bedrock_probe.rs (line 34)
4async fn main() {
5    let args: Vec<_> = env::args().collect();
6    let address = args[1].parse().unwrap();
7    let version: u8 = args[2].parse().unwrap();
8    rust_raknet::enable_raknet_log(3);
9    println!("TARGET {address} RAKNET {version}");
10    let socket = match tokio::time::timeout(
11        Duration::from_secs(10),
12        RaknetSocket::connect_with_version(&address, version),
13    )
14    .await
15    {
16        Ok(Ok(socket)) => {
17            println!("HANDSHAKE OK");
18            socket
19        }
20        Ok(Err(error)) => {
21            println!("HANDSHAKE ERROR {error}");
22            std::process::exit(1);
23        }
24        Err(_) => {
25            println!("HANDSHAKE TIMEOUT");
26            std::process::exit(2);
27        }
28    };
29    if let Some(protocol) = args.get(3) {
30        let protocol: i32 = protocol.parse().unwrap();
31        let mut request = vec![0xfe, 6, 0xc1, 1];
32        request.extend_from_slice(&protocol.to_be_bytes());
33        socket
34            .send(&request, Reliability::ReliableOrdered)
35            .await
36            .unwrap();
37        println!("SENT RequestNetworkSettings protocol={protocol}");
38        match tokio::time::timeout(Duration::from_secs(5), socket.recv()).await {
39            Ok(Ok(data)) => {
40                println!(
41                    "RECEIVED {} bytes: {:02x?}",
42                    data.len(),
43                    &data[..data.len().min(80)]
44                );
45                if data.len() >= 8 && data[0] == 0xfe && data[2..4] == [0x8f, 1] {
46                    println!(
47                        "NETWORK_SETTINGS OK threshold={} algorithm={}",
48                        u16::from_le_bytes([data[4], data[5]]),
49                        u16::from_le_bytes([data[6], data[7]])
50                    );
51                } else {
52                    println!("UNEXPECTED APPLICATION RESPONSE");
53                    std::process::exit(3);
54                }
55            }
56            other => {
57                println!("APPLICATION RECEIVE ERROR {other:?}");
58                std::process::exit(4);
59            }
60        }
61    }
62    let _ = tokio::time::timeout(Duration::from_secs(1), socket.close()).await;
63}
examples/interop.rs (line 45)
6async fn main() -> Result<(), Box<dyn Error>> {
7    if std::env::var_os("RAKNET_DEBUG").is_some() {
8        rust_raknet::enable_raknet_log(7);
9    }
10    let address = std::env::args()
11        .nth(1)
12        .ok_or("missing peer address")?
13        .parse()?;
14    let batch = std::env::args().nth(2).as_deref() == Some("--batch");
15    tokio::time::timeout(Duration::from_secs(60), async {
16        let socket = RaknetSocket::connect_with_version(&address, 11).await?;
17        let mut count = 0;
18        for size in [64, 800, 4096, 10000] {
19            if batch && size <= 800 {
20                for first in (0..100).step_by(16) {
21                    let messages: Vec<_> = (first..(first + 16).min(100))
22                        .map(|index| {
23                            let mut message: Vec<_> =
24                                (0..size).map(|i| (i + index) as u8).collect();
25                            message[0] = 0xfe;
26                            rust_raknet::Bytes::from(message)
27                        })
28                        .collect();
29                    socket
30                        .send_bytes_batch(&messages, Reliability::ReliableOrdered)
31                        .await?;
32                    for expected in messages {
33                        assert_eq!(
34                            socket.recv().await?.as_slice(),
35                            expected.as_ref(),
36                            "message reordered, duplicated or corrupted"
37                        );
38                        count += 1;
39                    }
40                }
41            } else {
42                for index in 0..100 {
43                    let mut payload: Vec<_> = (0..size).map(|i| (i + index) as u8).collect();
44                    payload[0] = 0xfe;
45                    socket.send(&payload, Reliability::ReliableOrdered).await?;
46                    assert_eq!(
47                        socket.recv().await?,
48                        payload,
49                        "message reordered, duplicated or corrupted"
50                    );
51                    count += 1;
52                }
53            }
54        }
55        socket.close().await?;
56        println!("PASS: {count} unique ordered messages, 64/800/4096/10000 bytes");
57        Ok::<_, Box<dyn Error>>(())
58    })
59    .await?
60}
examples/concurrency.rs (line 50)
11async fn main() -> Result<(), Box<dyn Error>> {
12    let args: Vec<_> = std::env::args().collect();
13    let address = args
14        .get(1)
15        .ok_or("usage: concurrency ADDRESS [CONNECTIONS] [MESSAGES]")?
16        .parse()?;
17    let connections: usize = args.get(2).map_or(Ok(512), |n| n.parse())?;
18    let messages: usize = args.get(3).map_or(Ok(100), |n| n.parse())?;
19    if connections == 0 || messages == 0 {
20        return Err("counts must be positive".into());
21    }
22    tokio::time::timeout(Duration::from_secs(120), async {
23        let started = Instant::now();
24        let dialing = Arc::new(Semaphore::new(8));
25        let mut tasks = JoinSet::new();
26        for index in 0..connections {
27            let dialing = dialing.clone();
28            tasks.spawn(async move {
29                let _permit = dialing.acquire().await.unwrap();
30                let socket = RaknetSocket::connect_with_version(&address, 11).await?;
31                Ok::<_, rust_raknet::error::RaknetError>((index, socket))
32            });
33        }
34        let mut sockets = Vec::with_capacity(connections);
35        while let Some(result) = tasks.join_next().await { sockets.push(result??); }
36        let handshake_seconds = started.elapsed().as_secs_f64();
37        let transfer = Instant::now();
38        let mut exchanges = JoinSet::new();
39        for (index, socket) in sockets {
40            exchanges.spawn(async move {
41                fn payload(connection: usize, message: usize) -> Vec<u8> {
42                    let size = [64, 800, 4096][message % 3];
43                    let mut bytes = vec![0xfe; size];
44                    bytes[1..9].copy_from_slice(&(connection as u64).to_le_bytes());
45                    bytes[9..17].copy_from_slice(&(message as u64).to_le_bytes());
46                    bytes
47                }
48                let sender = async {
49                    for message in 0..messages {
50                        socket.send(&payload(index, message), Reliability::ReliableOrdered).await?;
51                    }
52                    socket.flush().await
53                };
54                let receiver = async {
55                    // Exercise a slow application without blocking protocol ACKs.
56                    tokio::time::sleep(Duration::from_millis(30)).await;
57                    for message in 0..messages {
58                        let actual = socket.recv().await?;
59                        if actual != payload(index, message) {
60                            return Err(rust_raknet::error::RaknetError::PacketParseError);
61                        }
62                    }
63                    Ok(())
64                };
65                tokio::try_join!(sender, receiver)?;
66                socket.close().await
67            });
68        }
69        while let Some(result) = exchanges.join_next().await { result??; }
70        println!("PASS: {connections} simultaneous connections, {} unique ordered echoes; handshake={handshake_seconds:.3}s, transfer={:.3}s", connections * messages, transfer.elapsed().as_secs_f64());
71        Ok::<_, Box<dyn Error>>(())
72    }).await?
73}
Source

pub async fn send_with_order_channel( &self, buf: &[u8], reliability: Reliability, order_channel: u8, ) -> Result<()>

Send a packet on a specific RakNet ordering channel.

Each ordering channel has an independent reliable-ordered sequence. The channel is relevant to ordered and sequenced reliability modes; unordered modes do not carry an ordering channel on the wire.

Source

pub async fn send_bytes( &self, data: Bytes, reliability: Reliability, ) -> Result<()>

Send an owned, immutable payload without copying it into the retry queue.

Source

pub async fn send_bytes_with_order_channel( &self, data: Bytes, reliability: Reliability, order_channel: u8, ) -> Result<()>

Send an owned payload on a RakNet ordering channel.

Source

pub async fn send_batch( &self, messages: &[&[u8]], reliability: Reliability, ) -> Result<()>

Send an already available batch without waiting to collect more messages.

Small ReliableOrdered messages can share standard RakNet frame sets up to the negotiated MTU. Loss feedback or a retransmission timeout restores individual sends for this connection. Other modes retain individual datagrams. A batch is validated before sending any member. Cancellation can send a prefix; already queued reliable messages continue delivery. Empty batches are a no-op.

Source

pub async fn send_batch_with_order_channel( &self, messages: &[&[u8]], reliability: Reliability, order_channel: u8, ) -> Result<()>

Send a borrowed batch on one ordering channel.

Source

pub async fn send_bytes_batch( &self, messages: &[Bytes], reliability: Reliability, ) -> Result<()>

Send owned buffers without copying their payload into the retry queue.

Examples found in repository?
examples/sharded_echo.rs (line 33)
4async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
5    use rust_raknet::{RaknetListener, Reliability};
6    let address = std::env::args().nth(1).ok_or("ADDRESS required")?.parse()?;
7    let count = std::env::args()
8        .nth(2)
9        .unwrap_or_else(|| "4".into())
10        .parse()?;
11    let mut listener = RaknetListener::bind_with_socket_shards(&address, count).await?;
12    let mut batch_messages = false;
13    for option in std::env::args().skip(3) {
14        listener = match option.as_str() {
15            "--receive-batching" => listener.with_receive_batching(true),
16            "--idle-maintenance" => listener.with_idle_maintenance(true),
17            "--batch-messages" => {
18                batch_messages = true;
19                listener
20            }
21            _ => return Err(format!("unknown option: {option}").into()),
22        };
23    }
24    listener.listen().await;
25    println!("Sharded echo listening on {}", listener.local_addr()?);
26    loop {
27        let client = listener.accept().await?;
28        if batch_messages {
29            tokio::spawn(async move {
30                let mut messages = Vec::with_capacity(16);
31                while client.recv_bytes_batch(&mut messages, 16).await.is_ok() {
32                    if client
33                        .send_bytes_batch(&messages, Reliability::ReliableOrdered)
34                        .await
35                        .is_err()
36                    {
37                        break;
38                    }
39                    if messages.len() == 16 {
40                        tokio::task::yield_now().await;
41                    }
42                }
43            });
44        } else {
45            tokio::spawn(async move {
46                while let Ok(payload) = client.recv().await {
47                    if client
48                        .send(&payload, Reliability::ReliableOrdered)
49                        .await
50                        .is_err()
51                    {
52                        break;
53                    }
54                }
55            });
56        }
57    }
58}
More examples
Hide additional examples
examples/interop.rs (line 30)
6async fn main() -> Result<(), Box<dyn Error>> {
7    if std::env::var_os("RAKNET_DEBUG").is_some() {
8        rust_raknet::enable_raknet_log(7);
9    }
10    let address = std::env::args()
11        .nth(1)
12        .ok_or("missing peer address")?
13        .parse()?;
14    let batch = std::env::args().nth(2).as_deref() == Some("--batch");
15    tokio::time::timeout(Duration::from_secs(60), async {
16        let socket = RaknetSocket::connect_with_version(&address, 11).await?;
17        let mut count = 0;
18        for size in [64, 800, 4096, 10000] {
19            if batch && size <= 800 {
20                for first in (0..100).step_by(16) {
21                    let messages: Vec<_> = (first..(first + 16).min(100))
22                        .map(|index| {
23                            let mut message: Vec<_> =
24                                (0..size).map(|i| (i + index) as u8).collect();
25                            message[0] = 0xfe;
26                            rust_raknet::Bytes::from(message)
27                        })
28                        .collect();
29                    socket
30                        .send_bytes_batch(&messages, Reliability::ReliableOrdered)
31                        .await?;
32                    for expected in messages {
33                        assert_eq!(
34                            socket.recv().await?.as_slice(),
35                            expected.as_ref(),
36                            "message reordered, duplicated or corrupted"
37                        );
38                        count += 1;
39                    }
40                }
41            } else {
42                for index in 0..100 {
43                    let mut payload: Vec<_> = (0..size).map(|i| (i + index) as u8).collect();
44                    payload[0] = 0xfe;
45                    socket.send(&payload, Reliability::ReliableOrdered).await?;
46                    assert_eq!(
47                        socket.recv().await?,
48                        payload,
49                        "message reordered, duplicated or corrupted"
50                    );
51                    count += 1;
52                }
53            }
54        }
55        socket.close().await?;
56        println!("PASS: {count} unique ordered messages, 64/800/4096/10000 bytes");
57        Ok::<_, Box<dyn Error>>(())
58    })
59    .await?
60}
examples/concurrency_benchmark.rs (line 162)
79async fn run<const LOADED: bool>(
80    address: std::net::SocketAddr,
81    count: usize,
82    messages: usize,
83    size: usize,
84    tcp: bool,
85    batch: bool,
86) -> Result<(), BenchError> {
87    println!(
88        "Sending API: {}",
89        if tcp {
90            "TCP complete-record writes"
91        } else if batch {
92            "send_bytes_batch (ready messages only)"
93        } else {
94            "send (no batch API)"
95        }
96    );
97    let setup = Instant::now();
98    let dialing = Arc::new(Semaphore::new(8));
99    let mut tasks = JoinSet::new();
100    for id in 0..count {
101        let dialing = dialing.clone();
102        tasks.spawn(async move {
103            let _permit = dialing.acquire().await.unwrap();
104            let socket = EchoConnection::connect(address, tcp).await?;
105            Ok::<_, BenchError>((id, socket))
106        });
107    }
108    let mut sockets = Vec::with_capacity(count);
109    while let Some(result) = tasks.join_next().await {
110        sockets.push(result??);
111    }
112    let setup_seconds = setup.elapsed().as_secs_f64();
113    let barrier = Arc::new(Barrier::new(count + 1));
114    let start_gate = Arc::new(Barrier::new(count + 1));
115    let mut exchanges = JoinSet::new();
116    for (id, mut socket) in sockets {
117        let barrier = barrier.clone();
118        let start_gate = start_gate.clone();
119        exchanges.spawn(async move {
120            let mut rtts = Vec::with_capacity(if LOADED { messages.max(20) } else { 20 });
121            let mut stamps = if LOADED {
122                vec![Instant::now(); 16]
123            } else {
124                Vec::new()
125            };
126            let mut expected = payload(id, 0, size);
127            let mut outgoing = expected.clone();
128            let mut actual = Vec::with_capacity(size);
129            for message in 0..20 {
130                expected[9..17].copy_from_slice(&(message as u64).to_le_bytes());
131                let start = Instant::now();
132                socket.send(&expected).await?;
133                socket.recv(size, &mut actual).await?;
134                if actual != expected {
135                    return Err::<_, BenchError>("echo payload or order mismatch".into());
136                }
137                rtts.push(start.elapsed().as_nanos());
138            }
139            if LOADED {
140                rtts.clear();
141            }
142            barrier.wait().await;
143            start_gate.wait().await;
144            // Match the C reference's sliding application window. Refill a slot
145            // as each echo arrives instead of waiting for a whole batch.
146            if batch {
147                let EchoConnection::RakNet(ref connection) = socket else {
148                    unreachable!()
149                };
150                let mut sent = 0;
151                let mut received = 0;
152                let mut incoming = Vec::with_capacity(16);
153                let mut pending = Vec::with_capacity(16);
154                for message in 0..messages.min(16) {
155                    pending.push(rust_raknet::Bytes::from(payload(id, message + 20, size)));
156                    if LOADED {
157                        stamps[message % 16] = Instant::now();
158                    }
159                    sent += 1;
160                }
161                connection
162                    .send_bytes_batch(&pending, Reliability::ReliableOrdered)
163                    .await?;
164                while received < messages {
165                    connection.recv_bytes_batch(&mut incoming, 16).await?;
166                    for actual in &incoming {
167                        expected[9..17].copy_from_slice(&((received + 20) as u64).to_le_bytes());
168                        if actual.as_ref() != expected {
169                            return Err::<_, BenchError>("echo payload or order mismatch".into());
170                        }
171                        if LOADED {
172                            rtts.push(stamps[received % 16].elapsed().as_nanos());
173                        }
174                        received += 1;
175                    }
176                    pending.clear();
177                    for _ in 0..incoming.len().min(messages - sent) {
178                        pending.push(rust_raknet::Bytes::from(payload(id, sent + 20, size)));
179                        if LOADED {
180                            stamps[sent % 16] = Instant::now();
181                        }
182                        sent += 1;
183                    }
184                    if !pending.is_empty() {
185                        connection
186                            .send_bytes_batch(&pending, Reliability::ReliableOrdered)
187                            .await?;
188                    }
189                }
190            } else {
191                let mut sent = 0;
192                while sent < messages.min(16) {
193                    outgoing[9..17].copy_from_slice(&((sent + 20) as u64).to_le_bytes());
194                    if LOADED {
195                        stamps[sent % 16] = Instant::now();
196                    }
197                    socket.send(&outgoing).await?;
198                    sent += 1;
199                }
200                for received in 0..messages {
201                    expected[9..17].copy_from_slice(&((received + 20) as u64).to_le_bytes());
202                    socket.recv(size, &mut actual).await?;
203                    if actual != expected {
204                        return Err::<_, BenchError>("echo payload or order mismatch".into());
205                    }
206                    if LOADED {
207                        rtts.push(stamps[received % 16].elapsed().as_nanos());
208                    }
209                    if sent < messages {
210                        outgoing[9..17].copy_from_slice(&((sent + 20) as u64).to_le_bytes());
211                        if LOADED {
212                            stamps[sent % 16] = Instant::now();
213                        }
214                        socket.send(&outgoing).await?;
215                        sent += 1;
216                    }
217                }
218            }
219            // Keep every connection open until all peers finish the measured burst.
220            Ok::<_, BenchError>((socket, rtts))
221        });
222    }
223    barrier.wait().await;
224    let started = Instant::now();
225    start_gate.wait().await;
226    let mut retained = Vec::with_capacity(count);
227    let mut rtts = Vec::with_capacity(count * if LOADED { messages } else { 20 });
228    while let Some(result) = exchanges.join_next().await {
229        let (socket, values) = result??;
230        retained.push(socket);
231        rtts.extend(values);
232    }
233    let elapsed = started.elapsed().as_secs_f64();
234    println!("RTT samples: {}", rtts.len());
235    rtts.sort_unstable();
236    println!(
237        "Connections: {count}\nMessages per connection: {messages}\nPayload size: {size} bytes\nSetup: {setup_seconds:.6} s\nElapsed: {elapsed:.6} s"
238    );
239    println!(
240        "Echo payload throughput (per direction): {:.2} MiB/s",
241        count as f64 * messages as f64 * size as f64 / 1048576.0 / elapsed
242    );
243    println!(
244        "RTT p50: {:.1} us\nRTT p95: {:.1} us\nRTT p99: {:.1} us",
245        percentile(&rtts, 50),
246        percentile(&rtts, 95),
247        percentile(&rtts, 99)
248    );
249    println!(
250        "RTT measurement: {}",
251        if LOADED {
252            "per-message throughout measured burst"
253        } else {
254            "sequential warmup before burst"
255        }
256    );
257    println!("Verified ordered echoes: {}", count * (messages + 20));
258    drop(retained);
259    Ok(())
260}
Source

pub async fn send_bytes_batch_with_order_channel( &self, messages: &[Bytes], reliability: Reliability, order_channel: u8, ) -> Result<()>

Send an owned batch on one ordering channel.

Source

pub async fn flush(&self) -> Result<()>

Wait until all reliable packets have been acknowledged.

§Example
ⓘ
let socket = RaknetSocket::connect("127.0.0.1:19132".parse().unwrap()).await.unwrap();
socket.send(&[0xfe], Reliability::ReliableOrdered).await.unwrap();
socket.flush().await.unwrap();
Examples found in repository?
examples/concurrency.rs (line 52)
11async fn main() -> Result<(), Box<dyn Error>> {
12    let args: Vec<_> = std::env::args().collect();
13    let address = args
14        .get(1)
15        .ok_or("usage: concurrency ADDRESS [CONNECTIONS] [MESSAGES]")?
16        .parse()?;
17    let connections: usize = args.get(2).map_or(Ok(512), |n| n.parse())?;
18    let messages: usize = args.get(3).map_or(Ok(100), |n| n.parse())?;
19    if connections == 0 || messages == 0 {
20        return Err("counts must be positive".into());
21    }
22    tokio::time::timeout(Duration::from_secs(120), async {
23        let started = Instant::now();
24        let dialing = Arc::new(Semaphore::new(8));
25        let mut tasks = JoinSet::new();
26        for index in 0..connections {
27            let dialing = dialing.clone();
28            tasks.spawn(async move {
29                let _permit = dialing.acquire().await.unwrap();
30                let socket = RaknetSocket::connect_with_version(&address, 11).await?;
31                Ok::<_, rust_raknet::error::RaknetError>((index, socket))
32            });
33        }
34        let mut sockets = Vec::with_capacity(connections);
35        while let Some(result) = tasks.join_next().await { sockets.push(result??); }
36        let handshake_seconds = started.elapsed().as_secs_f64();
37        let transfer = Instant::now();
38        let mut exchanges = JoinSet::new();
39        for (index, socket) in sockets {
40            exchanges.spawn(async move {
41                fn payload(connection: usize, message: usize) -> Vec<u8> {
42                    let size = [64, 800, 4096][message % 3];
43                    let mut bytes = vec![0xfe; size];
44                    bytes[1..9].copy_from_slice(&(connection as u64).to_le_bytes());
45                    bytes[9..17].copy_from_slice(&(message as u64).to_le_bytes());
46                    bytes
47                }
48                let sender = async {
49                    for message in 0..messages {
50                        socket.send(&payload(index, message), Reliability::ReliableOrdered).await?;
51                    }
52                    socket.flush().await
53                };
54                let receiver = async {
55                    // Exercise a slow application without blocking protocol ACKs.
56                    tokio::time::sleep(Duration::from_millis(30)).await;
57                    for message in 0..messages {
58                        let actual = socket.recv().await?;
59                        if actual != payload(index, message) {
60                            return Err(rust_raknet::error::RaknetError::PacketParseError);
61                        }
62                    }
63                    Ok(())
64                };
65                tokio::try_join!(sender, receiver)?;
66                socket.close().await
67            });
68        }
69        while let Some(result) = exchanges.join_next().await { result??; }
70        println!("PASS: {connections} simultaneous connections, {} unique ordered echoes; handshake={handshake_seconds:.3}s, transfer={:.3}s", connections * messages, transfer.elapsed().as_secs_f64());
71        Ok::<_, Box<dyn Error>>(())
72    }).await?
73}
Source

pub async fn recv(&self) -> Result<Vec<u8>>

Receive the next game packet.

§Example
ⓘ
let socket = RaknetSocket::connect("127.0.0.1:19132".parse().unwrap()).await.unwrap();
let buf = socket.recv().await.unwrap();
if buf[0] == 0xfe{
   //do something
}
Examples found in repository?
examples/concurrency_benchmark.rs (line 301)
299    async fn recv(&mut self, size: usize, incoming: &mut Vec<u8>) -> Result<(), BenchError> {
300        match self {
301            Self::RakNet(socket) => *incoming = socket.recv().await?,
302            Self::Tcp { stream, .. } => {
303                let length = stream.read_u32_le().await? as usize;
304                if length != size {
305                    return Err("unexpected TCP record length".into());
306                }
307                incoming.resize(length, 0);
308                stream.read_exact(incoming).await?;
309            }
310        }
311        Ok(())
312    }
More examples
Hide additional examples
examples/sharded_echo.rs (line 46)
4async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
5    use rust_raknet::{RaknetListener, Reliability};
6    let address = std::env::args().nth(1).ok_or("ADDRESS required")?.parse()?;
7    let count = std::env::args()
8        .nth(2)
9        .unwrap_or_else(|| "4".into())
10        .parse()?;
11    let mut listener = RaknetListener::bind_with_socket_shards(&address, count).await?;
12    let mut batch_messages = false;
13    for option in std::env::args().skip(3) {
14        listener = match option.as_str() {
15            "--receive-batching" => listener.with_receive_batching(true),
16            "--idle-maintenance" => listener.with_idle_maintenance(true),
17            "--batch-messages" => {
18                batch_messages = true;
19                listener
20            }
21            _ => return Err(format!("unknown option: {option}").into()),
22        };
23    }
24    listener.listen().await;
25    println!("Sharded echo listening on {}", listener.local_addr()?);
26    loop {
27        let client = listener.accept().await?;
28        if batch_messages {
29            tokio::spawn(async move {
30                let mut messages = Vec::with_capacity(16);
31                while client.recv_bytes_batch(&mut messages, 16).await.is_ok() {
32                    if client
33                        .send_bytes_batch(&messages, Reliability::ReliableOrdered)
34                        .await
35                        .is_err()
36                    {
37                        break;
38                    }
39                    if messages.len() == 16 {
40                        tokio::task::yield_now().await;
41                    }
42                }
43            });
44        } else {
45            tokio::spawn(async move {
46                while let Ok(payload) = client.recv().await {
47                    if client
48                        .send(&payload, Reliability::ReliableOrdered)
49                        .await
50                        .is_err()
51                    {
52                        break;
53                    }
54                }
55            });
56        }
57    }
58}
examples/bedrock_probe.rs (line 38)
4async fn main() {
5    let args: Vec<_> = env::args().collect();
6    let address = args[1].parse().unwrap();
7    let version: u8 = args[2].parse().unwrap();
8    rust_raknet::enable_raknet_log(3);
9    println!("TARGET {address} RAKNET {version}");
10    let socket = match tokio::time::timeout(
11        Duration::from_secs(10),
12        RaknetSocket::connect_with_version(&address, version),
13    )
14    .await
15    {
16        Ok(Ok(socket)) => {
17            println!("HANDSHAKE OK");
18            socket
19        }
20        Ok(Err(error)) => {
21            println!("HANDSHAKE ERROR {error}");
22            std::process::exit(1);
23        }
24        Err(_) => {
25            println!("HANDSHAKE TIMEOUT");
26            std::process::exit(2);
27        }
28    };
29    if let Some(protocol) = args.get(3) {
30        let protocol: i32 = protocol.parse().unwrap();
31        let mut request = vec![0xfe, 6, 0xc1, 1];
32        request.extend_from_slice(&protocol.to_be_bytes());
33        socket
34            .send(&request, Reliability::ReliableOrdered)
35            .await
36            .unwrap();
37        println!("SENT RequestNetworkSettings protocol={protocol}");
38        match tokio::time::timeout(Duration::from_secs(5), socket.recv()).await {
39            Ok(Ok(data)) => {
40                println!(
41                    "RECEIVED {} bytes: {:02x?}",
42                    data.len(),
43                    &data[..data.len().min(80)]
44                );
45                if data.len() >= 8 && data[0] == 0xfe && data[2..4] == [0x8f, 1] {
46                    println!(
47                        "NETWORK_SETTINGS OK threshold={} algorithm={}",
48                        u16::from_le_bytes([data[4], data[5]]),
49                        u16::from_le_bytes([data[6], data[7]])
50                    );
51                } else {
52                    println!("UNEXPECTED APPLICATION RESPONSE");
53                    std::process::exit(3);
54                }
55            }
56            other => {
57                println!("APPLICATION RECEIVE ERROR {other:?}");
58                std::process::exit(4);
59            }
60        }
61    }
62    let _ = tokio::time::timeout(Duration::from_secs(1), socket.close()).await;
63}
examples/interop.rs (line 34)
6async fn main() -> Result<(), Box<dyn Error>> {
7    if std::env::var_os("RAKNET_DEBUG").is_some() {
8        rust_raknet::enable_raknet_log(7);
9    }
10    let address = std::env::args()
11        .nth(1)
12        .ok_or("missing peer address")?
13        .parse()?;
14    let batch = std::env::args().nth(2).as_deref() == Some("--batch");
15    tokio::time::timeout(Duration::from_secs(60), async {
16        let socket = RaknetSocket::connect_with_version(&address, 11).await?;
17        let mut count = 0;
18        for size in [64, 800, 4096, 10000] {
19            if batch && size <= 800 {
20                for first in (0..100).step_by(16) {
21                    let messages: Vec<_> = (first..(first + 16).min(100))
22                        .map(|index| {
23                            let mut message: Vec<_> =
24                                (0..size).map(|i| (i + index) as u8).collect();
25                            message[0] = 0xfe;
26                            rust_raknet::Bytes::from(message)
27                        })
28                        .collect();
29                    socket
30                        .send_bytes_batch(&messages, Reliability::ReliableOrdered)
31                        .await?;
32                    for expected in messages {
33                        assert_eq!(
34                            socket.recv().await?.as_slice(),
35                            expected.as_ref(),
36                            "message reordered, duplicated or corrupted"
37                        );
38                        count += 1;
39                    }
40                }
41            } else {
42                for index in 0..100 {
43                    let mut payload: Vec<_> = (0..size).map(|i| (i + index) as u8).collect();
44                    payload[0] = 0xfe;
45                    socket.send(&payload, Reliability::ReliableOrdered).await?;
46                    assert_eq!(
47                        socket.recv().await?,
48                        payload,
49                        "message reordered, duplicated or corrupted"
50                    );
51                    count += 1;
52                }
53            }
54        }
55        socket.close().await?;
56        println!("PASS: {count} unique ordered messages, 64/800/4096/10000 bytes");
57        Ok::<_, Box<dyn Error>>(())
58    })
59    .await?
60}
examples/concurrency.rs (line 58)
11async fn main() -> Result<(), Box<dyn Error>> {
12    let args: Vec<_> = std::env::args().collect();
13    let address = args
14        .get(1)
15        .ok_or("usage: concurrency ADDRESS [CONNECTIONS] [MESSAGES]")?
16        .parse()?;
17    let connections: usize = args.get(2).map_or(Ok(512), |n| n.parse())?;
18    let messages: usize = args.get(3).map_or(Ok(100), |n| n.parse())?;
19    if connections == 0 || messages == 0 {
20        return Err("counts must be positive".into());
21    }
22    tokio::time::timeout(Duration::from_secs(120), async {
23        let started = Instant::now();
24        let dialing = Arc::new(Semaphore::new(8));
25        let mut tasks = JoinSet::new();
26        for index in 0..connections {
27            let dialing = dialing.clone();
28            tasks.spawn(async move {
29                let _permit = dialing.acquire().await.unwrap();
30                let socket = RaknetSocket::connect_with_version(&address, 11).await?;
31                Ok::<_, rust_raknet::error::RaknetError>((index, socket))
32            });
33        }
34        let mut sockets = Vec::with_capacity(connections);
35        while let Some(result) = tasks.join_next().await { sockets.push(result??); }
36        let handshake_seconds = started.elapsed().as_secs_f64();
37        let transfer = Instant::now();
38        let mut exchanges = JoinSet::new();
39        for (index, socket) in sockets {
40            exchanges.spawn(async move {
41                fn payload(connection: usize, message: usize) -> Vec<u8> {
42                    let size = [64, 800, 4096][message % 3];
43                    let mut bytes = vec![0xfe; size];
44                    bytes[1..9].copy_from_slice(&(connection as u64).to_le_bytes());
45                    bytes[9..17].copy_from_slice(&(message as u64).to_le_bytes());
46                    bytes
47                }
48                let sender = async {
49                    for message in 0..messages {
50                        socket.send(&payload(index, message), Reliability::ReliableOrdered).await?;
51                    }
52                    socket.flush().await
53                };
54                let receiver = async {
55                    // Exercise a slow application without blocking protocol ACKs.
56                    tokio::time::sleep(Duration::from_millis(30)).await;
57                    for message in 0..messages {
58                        let actual = socket.recv().await?;
59                        if actual != payload(index, message) {
60                            return Err(rust_raknet::error::RaknetError::PacketParseError);
61                        }
62                    }
63                    Ok(())
64                };
65                tokio::try_join!(sender, receiver)?;
66                socket.close().await
67            });
68        }
69        while let Some(result) = exchanges.join_next().await { result??; }
70        println!("PASS: {connections} simultaneous connections, {} unique ordered echoes; handshake={handshake_seconds:.3}s, transfer={:.3}s", connections * messages, transfer.elapsed().as_secs_f64());
71        Ok::<_, Box<dyn Error>>(())
72    }).await?
73}
Source

pub async fn recv_bytes(&self) -> Result<Bytes>

Receive a game payload as an immutable buffer that can be forwarded with send_bytes without copying its application bytes.

Source

pub async fn recv_batch( &self, output: &mut Vec<Vec<u8>>, limit: usize, ) -> Result<usize>

Receive one message, then drain up to limit messages already available. Clears output and never waits to fill a batch. Work is capped at 64 messages per call; a zero limit returns immediately.

Source

pub async fn recv_bytes_batch( &self, output: &mut Vec<Bytes>, limit: usize, ) -> Result<usize>

Receive an immediately available batch for forwarding with send_bytes_batch.

Examples found in repository?
examples/sharded_echo.rs (line 31)
4async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
5    use rust_raknet::{RaknetListener, Reliability};
6    let address = std::env::args().nth(1).ok_or("ADDRESS required")?.parse()?;
7    let count = std::env::args()
8        .nth(2)
9        .unwrap_or_else(|| "4".into())
10        .parse()?;
11    let mut listener = RaknetListener::bind_with_socket_shards(&address, count).await?;
12    let mut batch_messages = false;
13    for option in std::env::args().skip(3) {
14        listener = match option.as_str() {
15            "--receive-batching" => listener.with_receive_batching(true),
16            "--idle-maintenance" => listener.with_idle_maintenance(true),
17            "--batch-messages" => {
18                batch_messages = true;
19                listener
20            }
21            _ => return Err(format!("unknown option: {option}").into()),
22        };
23    }
24    listener.listen().await;
25    println!("Sharded echo listening on {}", listener.local_addr()?);
26    loop {
27        let client = listener.accept().await?;
28        if batch_messages {
29            tokio::spawn(async move {
30                let mut messages = Vec::with_capacity(16);
31                while client.recv_bytes_batch(&mut messages, 16).await.is_ok() {
32                    if client
33                        .send_bytes_batch(&messages, Reliability::ReliableOrdered)
34                        .await
35                        .is_err()
36                    {
37                        break;
38                    }
39                    if messages.len() == 16 {
40                        tokio::task::yield_now().await;
41                    }
42                }
43            });
44        } else {
45            tokio::spawn(async move {
46                while let Ok(payload) = client.recv().await {
47                    if client
48                        .send(&payload, Reliability::ReliableOrdered)
49                        .await
50                        .is_err()
51                    {
52                        break;
53                    }
54                }
55            });
56        }
57    }
58}
More examples
Hide additional examples
examples/concurrency_benchmark.rs (line 165)
79async fn run<const LOADED: bool>(
80    address: std::net::SocketAddr,
81    count: usize,
82    messages: usize,
83    size: usize,
84    tcp: bool,
85    batch: bool,
86) -> Result<(), BenchError> {
87    println!(
88        "Sending API: {}",
89        if tcp {
90            "TCP complete-record writes"
91        } else if batch {
92            "send_bytes_batch (ready messages only)"
93        } else {
94            "send (no batch API)"
95        }
96    );
97    let setup = Instant::now();
98    let dialing = Arc::new(Semaphore::new(8));
99    let mut tasks = JoinSet::new();
100    for id in 0..count {
101        let dialing = dialing.clone();
102        tasks.spawn(async move {
103            let _permit = dialing.acquire().await.unwrap();
104            let socket = EchoConnection::connect(address, tcp).await?;
105            Ok::<_, BenchError>((id, socket))
106        });
107    }
108    let mut sockets = Vec::with_capacity(count);
109    while let Some(result) = tasks.join_next().await {
110        sockets.push(result??);
111    }
112    let setup_seconds = setup.elapsed().as_secs_f64();
113    let barrier = Arc::new(Barrier::new(count + 1));
114    let start_gate = Arc::new(Barrier::new(count + 1));
115    let mut exchanges = JoinSet::new();
116    for (id, mut socket) in sockets {
117        let barrier = barrier.clone();
118        let start_gate = start_gate.clone();
119        exchanges.spawn(async move {
120            let mut rtts = Vec::with_capacity(if LOADED { messages.max(20) } else { 20 });
121            let mut stamps = if LOADED {
122                vec![Instant::now(); 16]
123            } else {
124                Vec::new()
125            };
126            let mut expected = payload(id, 0, size);
127            let mut outgoing = expected.clone();
128            let mut actual = Vec::with_capacity(size);
129            for message in 0..20 {
130                expected[9..17].copy_from_slice(&(message as u64).to_le_bytes());
131                let start = Instant::now();
132                socket.send(&expected).await?;
133                socket.recv(size, &mut actual).await?;
134                if actual != expected {
135                    return Err::<_, BenchError>("echo payload or order mismatch".into());
136                }
137                rtts.push(start.elapsed().as_nanos());
138            }
139            if LOADED {
140                rtts.clear();
141            }
142            barrier.wait().await;
143            start_gate.wait().await;
144            // Match the C reference's sliding application window. Refill a slot
145            // as each echo arrives instead of waiting for a whole batch.
146            if batch {
147                let EchoConnection::RakNet(ref connection) = socket else {
148                    unreachable!()
149                };
150                let mut sent = 0;
151                let mut received = 0;
152                let mut incoming = Vec::with_capacity(16);
153                let mut pending = Vec::with_capacity(16);
154                for message in 0..messages.min(16) {
155                    pending.push(rust_raknet::Bytes::from(payload(id, message + 20, size)));
156                    if LOADED {
157                        stamps[message % 16] = Instant::now();
158                    }
159                    sent += 1;
160                }
161                connection
162                    .send_bytes_batch(&pending, Reliability::ReliableOrdered)
163                    .await?;
164                while received < messages {
165                    connection.recv_bytes_batch(&mut incoming, 16).await?;
166                    for actual in &incoming {
167                        expected[9..17].copy_from_slice(&((received + 20) as u64).to_le_bytes());
168                        if actual.as_ref() != expected {
169                            return Err::<_, BenchError>("echo payload or order mismatch".into());
170                        }
171                        if LOADED {
172                            rtts.push(stamps[received % 16].elapsed().as_nanos());
173                        }
174                        received += 1;
175                    }
176                    pending.clear();
177                    for _ in 0..incoming.len().min(messages - sent) {
178                        pending.push(rust_raknet::Bytes::from(payload(id, sent + 20, size)));
179                        if LOADED {
180                            stamps[sent % 16] = Instant::now();
181                        }
182                        sent += 1;
183                    }
184                    if !pending.is_empty() {
185                        connection
186                            .send_bytes_batch(&pending, Reliability::ReliableOrdered)
187                            .await?;
188                    }
189                }
190            } else {
191                let mut sent = 0;
192                while sent < messages.min(16) {
193                    outgoing[9..17].copy_from_slice(&((sent + 20) as u64).to_le_bytes());
194                    if LOADED {
195                        stamps[sent % 16] = Instant::now();
196                    }
197                    socket.send(&outgoing).await?;
198                    sent += 1;
199                }
200                for received in 0..messages {
201                    expected[9..17].copy_from_slice(&((received + 20) as u64).to_le_bytes());
202                    socket.recv(size, &mut actual).await?;
203                    if actual != expected {
204                        return Err::<_, BenchError>("echo payload or order mismatch".into());
205                    }
206                    if LOADED {
207                        rtts.push(stamps[received % 16].elapsed().as_nanos());
208                    }
209                    if sent < messages {
210                        outgoing[9..17].copy_from_slice(&((sent + 20) as u64).to_le_bytes());
211                        if LOADED {
212                            stamps[sent % 16] = Instant::now();
213                        }
214                        socket.send(&outgoing).await?;
215                        sent += 1;
216                    }
217                }
218            }
219            // Keep every connection open until all peers finish the measured burst.
220            Ok::<_, BenchError>((socket, rtts))
221        });
222    }
223    barrier.wait().await;
224    let started = Instant::now();
225    start_gate.wait().await;
226    let mut retained = Vec::with_capacity(count);
227    let mut rtts = Vec::with_capacity(count * if LOADED { messages } else { 20 });
228    while let Some(result) = exchanges.join_next().await {
229        let (socket, values) = result??;
230        retained.push(socket);
231        rtts.extend(values);
232    }
233    let elapsed = started.elapsed().as_secs_f64();
234    println!("RTT samples: {}", rtts.len());
235    rtts.sort_unstable();
236    println!(
237        "Connections: {count}\nMessages per connection: {messages}\nPayload size: {size} bytes\nSetup: {setup_seconds:.6} s\nElapsed: {elapsed:.6} s"
238    );
239    println!(
240        "Echo payload throughput (per direction): {:.2} MiB/s",
241        count as f64 * messages as f64 * size as f64 / 1048576.0 / elapsed
242    );
243    println!(
244        "RTT p50: {:.1} us\nRTT p95: {:.1} us\nRTT p99: {:.1} us",
245        percentile(&rtts, 50),
246        percentile(&rtts, 95),
247        percentile(&rtts, 99)
248    );
249    println!(
250        "RTT measurement: {}",
251        if LOADED {
252            "per-message throughout measured burst"
253        } else {
254            "sequential warmup before burst"
255        }
256    );
257    println!("Verified ordered echoes: {}", count * (messages + 20));
258    drop(retained);
259    Ok(())
260}
Source

pub fn peer_addr(&self) -> Result<SocketAddr>

Return the remote peer’s UDP address.

§Example
ⓘ
let socket = RaknetSocket::connect("127.0.0.1:19132".parse().unwrap()).await.unwrap();
assert_eq!(socket.peer_addr().unwrap(), SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(127, 0, 0, 1), 19132)));
Source

pub fn local_addr(&self) -> Result<SocketAddr>

Return the local UDP address used by this connection.

§Example
ⓘ
let mut socket = RaknetSocket::connect("127.0.0.1:19132".parse().unwrap()).await.unwrap();
assert_eq!(socket.local_addr().unwrap().ip(), IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)));
Source

pub fn raknet_version(&self) -> Result<u8>

Return the RakNet protocol version used by this connection.

Source

pub fn set_loss_rate(&mut self, stage: u8)

Enable simulated packet loss for testing.

stage ranges from 0 (no loss) to 10 (all packets dropped).

§Example
ⓘ
let mut socket = RaknetSocket::connect("127.0.0.1:19132".parse().unwrap()).await.unwrap();
// Simulate 20% packet loss.
socket.set_loss_rate(2);

Trait Implementations§

Source§

impl Drop for RaknetSocket

Source§

fn drop(&mut self)

Executes the destructor for this type. Read more
Source§

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)
Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned. Read more

Auto Trait Implementations§

Blanket Implementations§

Source§

impl<T> Any for T
where T: 'static + ?Sized,

Source§

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more
Source§

impl<T> Borrow<T> for T
where T: ?Sized,

Source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more
Source§

impl<T> BorrowMut<T> for T
where T: ?Sized,

Source§

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more
Source§

impl<ST, DT> CastableFrom<ST, Initialized, Initialized> for DT
where ST: ?Sized, DT: ?Sized,

Source§

impl<ST, DT> CastableFrom<ST, Uninit, Uninit> for DT
where ST: ?Sized, DT: ?Sized,

Source§

impl<T> From<T> for T

Source§

fn from(t: T) -> T

Returns the argument unchanged.

Source§

impl<T, U> Into<U> for T
where U: From<T>,

Source§

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

Source§

impl<T> Read<Exclusive, BecauseExclusive> for T
where T: ?Sized,

Source§

impl<T, U> TryFrom<U> for T
where U: Into<T>,

Source§

type Error = !

The type returned in the event of a conversion error.
Source§

fn try_from(value: U) -> Result<T, !>

Performs the conversion.
Source§

impl<T, U> TryInto<U> for T
where U: TryFrom<T>,

Source§

type Error = <U as TryFrom<T>>::Error

The type returned in the event of a conversion error.
Source§

fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>

Performs the conversion.
Source§

impl<V, T> VZip<V> for T
where V: MultiLane<T>,

Source§

fn vzip(self) -> V