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 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 177)
167    async fn connect(address: std::net::SocketAddr, tcp: bool) -> Result<Self, BenchError> {
168        if tcp {
169            let stream = TcpStream::connect(address).await?;
170            stream.set_nodelay(true)?;
171            Ok(Self::Tcp {
172                stream,
173                record: Vec::new(),
174            })
175        } else {
176            Ok(Self::RakNet(
177                RaknetSocket::connect_with_version(&address, 11).await?,
178            ))
179        }
180    }
More examples
Hide additional examples
examples/interop.rs (line 15)
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    tokio::time::timeout(Duration::from_secs(60), async {
15        let socket = RaknetSocket::connect_with_version(&address, 11).await?;
16        let mut count = 0;
17        for size in [64, 800, 4096, 10000] {
18            for index in 0..100 {
19                let mut payload: Vec<_> = (0..size).map(|i| (i + index) as u8).collect();
20                payload[0] = 0xfe;
21                socket.send(&payload, Reliability::ReliableOrdered).await?;
22                assert_eq!(
23                    socket.recv().await?,
24                    payload,
25                    "message reordered, duplicated or corrupted"
26                );
27                count += 1;
28            }
29        }
30        socket.close().await?;
31        println!("PASS: {count} unique ordered messages, 64/800/4096/10000 bytes");
32        Ok::<_, Box<dyn Error>>(())
33    })
34    .await?
35}
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/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/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    tokio::time::timeout(Duration::from_secs(60), async {
15        let socket = RaknetSocket::connect_with_version(&address, 11).await?;
16        let mut count = 0;
17        for size in [64, 800, 4096, 10000] {
18            for index in 0..100 {
19                let mut payload: Vec<_> = (0..size).map(|i| (i + index) as u8).collect();
20                payload[0] = 0xfe;
21                socket.send(&payload, Reliability::ReliableOrdered).await?;
22                assert_eq!(
23                    socket.recv().await?,
24                    payload,
25                    "message reordered, duplicated or corrupted"
26                );
27                count += 1;
28            }
29        }
30        socket.close().await?;
31        println!("PASS: {count} unique ordered messages, 64/800/4096/10000 bytes");
32        Ok::<_, Box<dyn Error>>(())
33    })
34    .await?
35}
More examples
Hide additional examples
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}
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 184)
182    async fn send(&mut self, payload: &[u8]) -> Result<(), BenchError> {
183        match self {
184            Self::RakNet(socket) => socket.send(payload, Reliability::ReliableOrdered).await?,
185            Self::Tcp { stream, record, .. } => {
186                let length = u32::try_from(payload.len())?;
187                record.clear();
188                record.extend_from_slice(&length.to_le_bytes());
189                record.extend_from_slice(payload);
190                stream.write_all(record).await?;
191            }
192        }
193        Ok(())
194    }
More examples
Hide additional examples
examples/sharded_echo.rs (line 19)
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    listener.listen().await;
13    println!("Sharded echo listening on {}", listener.local_addr()?);
14    loop {
15        let client = listener.accept().await?;
16        tokio::spawn(async move {
17            while let Ok(payload) = client.recv().await {
18                if client
19                    .send(&payload, Reliability::ReliableOrdered)
20                    .await
21                    .is_err()
22                {
23                    break;
24                }
25            }
26        });
27    }
28}
examples/interop.rs (line 21)
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    tokio::time::timeout(Duration::from_secs(60), async {
15        let socket = RaknetSocket::connect_with_version(&address, 11).await?;
16        let mut count = 0;
17        for size in [64, 800, 4096, 10000] {
18            for index in 0..100 {
19                let mut payload: Vec<_> = (0..size).map(|i| (i + index) as u8).collect();
20                payload[0] = 0xfe;
21                socket.send(&payload, Reliability::ReliableOrdered).await?;
22                assert_eq!(
23                    socket.recv().await?,
24                    payload,
25                    "message reordered, duplicated or corrupted"
26                );
27                count += 1;
28            }
29        }
30        socket.close().await?;
31        println!("PASS: {count} unique ordered messages, 64/800/4096/10000 bytes");
32        Ok::<_, Box<dyn Error>>(())
33    })
34    .await?
35}
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/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 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 198)
196    async fn recv(&mut self, size: usize, incoming: &mut Vec<u8>) -> Result<(), BenchError> {
197        match self {
198            Self::RakNet(socket) => *incoming = socket.recv().await?,
199            Self::Tcp { stream, .. } => {
200                let length = stream.read_u32_le().await? as usize;
201                if length != size {
202                    return Err("unexpected TCP record length".into());
203                }
204                incoming.resize(length, 0);
205                stream.read_exact(incoming).await?;
206            }
207        }
208        Ok(())
209    }
More examples
Hide additional examples
examples/sharded_echo.rs (line 17)
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    listener.listen().await;
13    println!("Sharded echo listening on {}", listener.local_addr()?);
14    loop {
15        let client = listener.accept().await?;
16        tokio::spawn(async move {
17            while let Ok(payload) = client.recv().await {
18                if client
19                    .send(&payload, Reliability::ReliableOrdered)
20                    .await
21                    .is_err()
22                {
23                    break;
24                }
25            }
26        });
27    }
28}
examples/interop.rs (line 23)
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    tokio::time::timeout(Duration::from_secs(60), async {
15        let socket = RaknetSocket::connect_with_version(&address, 11).await?;
16        let mut count = 0;
17        for size in [64, 800, 4096, 10000] {
18            for index in 0..100 {
19                let mut payload: Vec<_> = (0..size).map(|i| (i + index) as u8).collect();
20                payload[0] = 0xfe;
21                socket.send(&payload, Reliability::ReliableOrdered).await?;
22                assert_eq!(
23                    socket.recv().await?,
24                    payload,
25                    "message reordered, duplicated or corrupted"
26                );
27                count += 1;
28            }
29        }
30        socket.close().await?;
31        println!("PASS: {count} unique ordered messages, 64/800/4096/10000 bytes");
32        Ok::<_, Box<dyn Error>>(())
33    })
34    .await?
35}
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/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 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