moirai-async 0.11.3

Async/await support for Moirai concurrency library
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

use moirai_pal::net::{AsyncUdpSocket, UdpBufferSizes, UdpConfig};
use std::io;
use std::net::SocketAddr;
use std::sync::Arc;

use crate::net::types::{UdpSocketStats, UdpStats};

/// Native async UDP socket
pub struct UdpSocket {
    inner: AsyncUdpSocket,
    stats: Arc<UdpStats>,
    config: UdpConfig,
}

#[cfg(test)]
mod tests;

impl UdpSocket {
    /// Bind UDP socket to an address
    pub async fn bind(addr: &str) -> io::Result<Self> {
        Self::bind_with_config(addr, UdpConfig::default()).await
    }

    /// Bind UDP socket after applying all requested native options.
    ///
    /// Invalid options or a failed native operation return an error before the
    /// socket is published. Absent buffer requests retain kernel defaults.
    pub async fn bind_with_config(addr: &str, config: UdpConfig) -> io::Result<Self> {
        let addr_parsed = crate::net::resolve::resolve(addr).await?.first();
        let inner = AsyncUdpSocket::bind_with_config(addr_parsed, &config).await?;

        Ok(Self {
            inner,
            stats: Arc::new(UdpStats::default()),
            config,
        })
    }

    /// Send data to a specific address
    pub async fn send_to(&self, buf: &[u8], target: SocketAddr) -> io::Result<usize> {
        let bytes_sent = self.inner.send_to(buf, target).await?;
        self.stats
            .packets_sent
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        self.stats
            .bytes_sent
            .fetch_add(bytes_sent as u64, std::sync::atomic::Ordering::Relaxed);
        Ok(bytes_sent)
    }

    /// Receive data from any address
    pub async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
        let (bytes_received, addr) = self.inner.recv_from(buf).await?;
        self.stats
            .packets_received
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        self.stats
            .bytes_received
            .fetch_add(bytes_received as u64, std::sync::atomic::Ordering::Relaxed);
        Ok((bytes_received, addr))
    }

    /// Get UDP socket statistics
    pub fn stats(&self) -> UdpSocketStats {
        UdpSocketStats {
            packets_sent: self
                .stats
                .packets_sent
                .load(std::sync::atomic::Ordering::Relaxed),
            packets_received: self
                .stats
                .packets_received
                .load(std::sync::atomic::Ordering::Relaxed),
            bytes_sent: self
                .stats
                .bytes_sent
                .load(std::sync::atomic::Ordering::Relaxed),
            bytes_received: self
                .stats
                .bytes_received
                .load(std::sync::atomic::Ordering::Relaxed),
        }
    }

    /// Return the local socket address this UDP socket is bound to.
    pub fn local_addr(&self) -> io::Result<SocketAddr> {
        self.inner.local_addr()
    }

    /// Return the socket configuration used at bind time.
    pub fn config(&self) -> &UdpConfig {
        &self.config
    }

    /// Read effective native receive/send buffers, including kernel accounting.
    pub fn buffer_sizes(&self) -> io::Result<UdpBufferSizes> {
        self.inner.buffer_sizes()
    }
}