libutp-rs 1.0.0

Async libutp interface
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
100
101
102
103
104
105
106
107
108
109
110
111
112
113

use crate::thread::{StopFlag, ThreadRunner};

use mio::{Events, Interest, Token};
use std::io;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;

/// An event loop for a mio::net::UdpSocket
///
/// Operates using three closures:
///
/// on_read_ready:   called when the socket is ready to read, and should return a bool indicating if
///                  the socket was read until WouldBlock
/// on_write_ready:  called when the socket is ready to write
/// on_err:          called if an error occurs
struct EventLoop<R, W, E> {
    io: Arc<mio::net::UdpSocket>,
    poll: mio::Poll,
    on_read_ready: R,
    on_write_ready: W,
    on_err: E,
}

impl<R, W, E> EventLoop<R, W, E>
where
    R: FnMut(&mio::net::UdpSocket) -> bool + 'static + Send,
    W: FnMut(&mio::net::UdpSocket) + 'static + Send,
    E: FnMut(io::Error) + 'static + Send,
{
    fn new(
        io: &mut Arc<mio::net::UdpSocket>,
        on_read_ready: R,
        on_write_ready: W,
        on_err: E,
    ) -> io::Result<EventLoop<R, W, E>> {
        let poll = mio::Poll::new()?;
        poll.registry().register(
            Arc::get_mut(io).unwrap(),
            Token(0),
            Interest::READABLE | Interest::WRITABLE,
        )?;
        let io = Arc::clone(&io);
        let event_loop = EventLoop {
            io,
            poll,
            on_read_ready,
            on_write_ready,
            on_err,
        };
        Ok(event_loop)
    }

    fn run(&mut self, stop_flag: StopFlag) {
        let mut events = Events::with_capacity(128);
        let mut read_ready = false;
        let mut send_ready = false;
        while !stop_flag.is_set() {
            if let Err(e) = self
                .poll
                .poll(&mut events, Some(Duration::from_millis(100)))
            {
                (self.on_err)(e)
            }

            for event in events.iter() {
                read_ready = event.is_readable();
                send_ready = event.is_writable();
            }

            if read_ready {
                read_ready = !(self.on_read_ready)(&self.io);
            }

            if send_ready {
                (self.on_write_ready)(&self.io);
                send_ready = false;
            }
        }
    }
}

pub(crate) struct UdpSocket {
    io: Arc<mio::net::UdpSocket>,
    _event_loop: ThreadRunner,
}

impl UdpSocket {
    /// Binds a udp socket to addr and spawns a thread running an event loop that is
    /// responsible for waking the receiver and senders when they are ready to send/receive
    pub(crate) fn bind<R, W, E>(
        addr: SocketAddr,
        on_read_ready: R,
        on_write_ready: W,
        on_err: E,
    ) -> io::Result<UdpSocket>
    where
        R: FnMut(&mio::net::UdpSocket) -> bool + 'static + Send,
        W: FnMut(&mio::net::UdpSocket) + 'static + Send,
        E: FnMut(io::Error) + 'static + Send,
    {
        let mut io = Arc::new(mio::net::UdpSocket::bind(addr)?);
        let mut _event_loop = EventLoop::new(&mut io, on_read_ready, on_write_ready, on_err)?;
        Ok(UdpSocket {
            io,
            _event_loop: ThreadRunner::run(move |stop_flag| _event_loop.run(stop_flag)),
        })
    }

    pub(crate) fn inner(&self) -> &Arc<mio::net::UdpSocket> {
        &self.io
    }
}