gaea/net/udp.rs
1use std::io;
2use std::net::SocketAddr;
3#[cfg(unix)]
4use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
5
6use crate::os::{Evented, Interests, OsQueue, RegisterOption};
7use crate::{event, sys};
8
9/// A User Datagram Protocol socket.
10///
11/// This works much like the `UdpSocket` in the standard library, but the I/O
12/// methods such as [`send_to`], [`send`] etc. don't block and instead return a
13/// [`WouldBlock`] error.
14///
15/// [`send_to`]: UdpSocket::send_to
16/// [`send`]: UdpSocket::send
17/// [`WouldBlock`]: std::io::ErrorKind::WouldBlock
18///
19/// # Deregistering
20///
21/// `UdpSocket` will deregister itself when dropped.
22///
23/// # Examples
24///
25/// An simple echo program, the `sender` sends a message and the `echoer`
26/// listens for messages and prints them to standard out.
27///
28/// ```
29/// # fn main() -> Result<(), Box<dyn std::error::Error>> {
30/// use std::io;
31///
32/// use gaea::{event, poll};
33/// use gaea::net::UdpSocket;
34/// use gaea::os::{Interests, RegisterOption, OsQueue};
35///
36/// // Unique ids and addresses for both the sender and echoer.
37/// const SENDER_ID: event::Id = event::Id(0);
38/// const ECHOER_ID: event::Id = event::Id(1);
39///
40/// let sender_address = "127.0.0.1:7000".parse()?;
41/// let echoer_address = "127.0.0.1:7001".parse()?;
42///
43/// // Create our sockets.
44/// let mut sender_socket = UdpSocket::bind(sender_address)?;
45/// let mut echoer_socket = UdpSocket::bind(echoer_address)?;
46///
47/// // Connect the sending socket so we can use `send` method.
48/// sender_socket.connect(echoer_address)?;
49///
50/// // As always create our poll and events.
51/// let mut os_queue = OsQueue::new()?;
52/// let mut events = Vec::new();
53///
54/// // Register our sockets
55/// os_queue.register(&mut sender_socket, SENDER_ID, Interests::WRITABLE, RegisterOption::LEVEL)?;
56/// os_queue.register(&mut echoer_socket, ECHOER_ID, Interests::READABLE, RegisterOption::LEVEL)?;
57///
58/// // The message we'll send.
59/// const MSG_TO_SEND: &[u8; 11] = b"Hello world";
60/// // A buffer for our echoer to receive the message in.
61/// let mut buf = [0; 20];
62///
63/// // Our event loop.
64/// loop {
65/// // Poll for events.
66/// poll::<_, io::Error>(&mut [&mut os_queue], &mut events, None)?;
67///
68/// for event in &mut events {
69/// match event.id() {
70/// SENDER_ID => {
71/// // Our sender is ready to send.
72/// let bytes_sent = sender_socket.send(MSG_TO_SEND)?;
73/// assert_eq!(bytes_sent, MSG_TO_SEND.len());
74/// println!("sent {:?} ({} bytes)", MSG_TO_SEND, bytes_sent);
75/// },
76/// ECHOER_ID => {
77/// // Our echoer is ready to read.
78/// let (bytes_recv, address) = echoer_socket.recv_from(&mut buf)?;
79/// println!("received {:?} ({} bytes) from {}", &buf[0..bytes_recv], bytes_recv, address);
80/// # return Ok(());
81/// }
82/// // We shouldn't receive any event with another id then the two
83/// // defined above.
84/// _ => unreachable!("received an unexpected event")
85/// }
86/// }
87/// }
88/// # }
89/// ```
90#[derive(Debug)]
91pub struct UdpSocket {
92 socket: sys::UdpSocket,
93}
94
95impl UdpSocket {
96 /// The interests to use when registering to receive both readable and
97 /// writable events.
98 pub const INTERESTS: Interests = Interests::BOTH;
99
100 /// Creates a UDP socket and binds it to the given address.
101 ///
102 /// # Examples
103 ///
104 /// ```
105 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
106 /// use gaea::net::UdpSocket;
107 ///
108 /// // We must bind it to an open address.
109 /// let address = "127.0.0.1:7002".parse()?;
110 /// let socket = UdpSocket::bind(address)?;
111 ///
112 /// // Our socket was created, but we should not use it before checking it's
113 /// // readiness.
114 /// # drop(socket); // Silence unused variable warning.
115 /// # Ok(())
116 /// # }
117 /// ```
118 pub fn bind(address: SocketAddr) -> io::Result<UdpSocket> {
119 sys::UdpSocket::bind(address).map(|socket| UdpSocket { socket })
120 }
121
122 /// Connects the UDP socket by setting the default destination and limiting
123 /// packets that are read, written and peeked to the address specified in
124 /// `address`.
125 ///
126 /// This allows the [`send`], [`recv`] and [`peek`] methods to be used.
127 ///
128 /// [`send`]: UdpSocket::send
129 /// [`recv`]: UdpSocket::recv
130 /// [`peek`]: UdpSocket::peek
131 pub fn connect(&mut self, address: SocketAddr) -> io::Result<()> {
132 self.socket.connect(address)
133 }
134
135 /// Returns the socket address that this socket was created from.
136 ///
137 /// # Examples
138 ///
139 /// ```
140 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
141 /// use gaea::net::UdpSocket;
142 ///
143 /// let address = "127.0.0.1:7003".parse()?;
144 /// let mut socket = UdpSocket::bind(address)?;
145 ///
146 /// assert_eq!(socket.local_addr()?, address);
147 /// # Ok(())
148 /// # }
149 pub fn local_addr(&mut self) -> io::Result<SocketAddr> {
150 self.socket.local_addr()
151 }
152
153 /// Sends data to the given address. On success, returns the number of bytes
154 /// written.
155 ///
156 /// # Examples
157 ///
158 /// ```
159 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
160 /// use std::io;
161 ///
162 /// use gaea::net::UdpSocket;
163 /// use gaea::os::{RegisterOption, Interests};
164 /// use gaea::{event, OsQueue, poll};
165 ///
166 /// let mut os_queue = OsQueue::new()?;
167 /// let mut events = Vec::new();
168 ///
169 /// let address = "127.0.0.1:7004".parse()?;
170 /// let mut socket = UdpSocket::bind(address)?;
171 ///
172 /// // Register our socket.
173 /// os_queue.register(&mut socket, event::Id(0), Interests::WRITABLE, RegisterOption::EDGE)?;
174 ///
175 /// // Poll until our socket is ready.
176 /// while events.is_empty() {
177 /// poll::<_, io::Error>(&mut [&mut os_queue], &mut events, None)?;
178 /// }
179 ///
180 /// let other_address = "127.0.0.1:7005".parse()?;
181 /// let bytes_sent = socket.send_to(b"Hello world", other_address)?;
182 /// assert_eq!(bytes_sent, 11);
183 /// #
184 /// # Ok(())
185 /// # }
186 /// ```
187 pub fn send_to(&mut self, buf: &[u8], target: SocketAddr) -> io::Result<usize> {
188 self.socket.send_to(buf, &target)
189 }
190
191 /// Sends data on the socket to the connected socket. On success, returns
192 /// the number of bytes written.
193 ///
194 /// # Examples
195 ///
196 /// ```
197 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
198 /// use std::io;
199 ///
200 /// use gaea::net::UdpSocket;
201 /// use gaea::os::{RegisterOption, Interests};
202 /// use gaea::{event, OsQueue, poll};
203 ///
204 /// let mut os_queue = OsQueue::new()?;
205 /// let mut events = Vec::new();
206 ///
207 /// let local_address = "127.0.0.1:7006".parse()?;
208 /// let remote_address = "127.0.0.1:7007".parse()?;
209 /// let mut socket = UdpSocket::bind(local_address)?;
210 /// socket.connect(remote_address)?;
211 ///
212 /// // Register our socket.
213 /// os_queue.register(&mut socket, event::Id(0), Interests::WRITABLE, RegisterOption::EDGE)?;
214 ///
215 /// // Poll until our socket is ready.
216 /// while events.is_empty() {
217 /// poll::<_, io::Error>(&mut [&mut os_queue], &mut events, None)?;
218 /// }
219 ///
220 /// let bytes_sent = socket.send(b"Hello world")?;
221 /// assert_eq!(bytes_sent, 11);
222 /// #
223 /// # Ok(())
224 /// # }
225 /// ```
226 ///
227 /// # Notes
228 ///
229 /// This requires the socket to be [connected].
230 ///
231 /// [connected]: UdpSocket::connect
232 pub fn send(&mut self, buf: &[u8]) -> io::Result<usize> {
233 self.socket.send(buf)
234 }
235
236 /// Receives data from the socket. On success, returns the number of bytes
237 /// read and the address from whence the data came.
238 ///
239 /// # Examples
240 ///
241 /// ```
242 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
243 /// use std::io;
244 ///
245 /// use gaea::net::UdpSocket;
246 /// use gaea::os::{RegisterOption, Interests};
247 /// use gaea::{event, OsQueue, poll};
248 ///
249 /// let mut os_queue = OsQueue::new()?;
250 /// let mut events = Vec::new();
251 ///
252 /// let address = "127.0.0.1:7008".parse()?;
253 /// let mut socket = UdpSocket::bind(address)?;
254 /// #
255 /// # // Send some data that we can receive.
256 /// # let mut socket2 = UdpSocket::bind("127.0.0.1:7108".parse()?)?;
257 /// # os_queue.register(&mut socket2, event::Id(1), Interests::WRITABLE, RegisterOption::EDGE)?;
258 /// # while events.is_empty() { poll::<_, io::Error>(&mut [&mut os_queue], &mut events, None)?; }
259 /// # let bytes_sent = socket2.send_to(b"Hello world", address)?;
260 /// # assert_eq!(bytes_sent, 11);
261 /// # events.clear();
262 ///
263 /// // Register our socket.
264 /// os_queue.register(&mut socket, event::Id(0), Interests::READABLE, RegisterOption::EDGE)?;
265 ///
266 /// // Poll until our socket is ready.
267 /// while events.is_empty() {
268 /// poll::<_, io::Error>(&mut [&mut os_queue], &mut events, None)?;
269 /// }
270 ///
271 /// let mut buf = [0; 20];
272 /// let (bytes_received, from_address) = socket.recv_from(&mut buf)?;
273 /// println!("Received {:?} ({} bytes) from {}", &buf[..bytes_received], bytes_received, from_address);
274 /// # assert_eq!(&buf[..bytes_received], b"Hello world");
275 /// #
276 /// # Ok(())
277 /// # }
278 /// ```
279 pub fn recv_from(&mut self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
280 self.socket.recv_from(buf)
281 }
282
283 /// Receives data from the socket. On success, returns the number of bytes
284 /// read.
285 ///
286 /// # Examples
287 ///
288 /// ```
289 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
290 /// use std::io;
291 ///
292 /// use gaea::net::UdpSocket;
293 /// use gaea::os::{RegisterOption, Interests};
294 /// use gaea::{event, OsQueue, poll};
295 ///
296 /// let mut os_queue = OsQueue::new()?;
297 /// let mut events = Vec::new();
298 ///
299 /// let local_address = "127.0.0.1:7009".parse()?;
300 /// let remote_address = "127.0.0.1:7010".parse()?;
301 /// let mut socket = UdpSocket::bind(local_address)?;
302 /// #
303 /// # // Send some data that we can receive.
304 /// # let mut socket2 = UdpSocket::bind(remote_address)?;
305 /// # os_queue.register(&mut socket2, event::Id(1), Interests::WRITABLE, RegisterOption::EDGE)?;
306 /// # while events.is_empty() { poll::<_, io::Error>(&mut [&mut os_queue], &mut events, None)?; }
307 /// # let bytes_sent = socket2.send_to(b"Hello world", local_address)?;
308 /// # assert_eq!(bytes_sent, 11);
309 /// # events.clear();
310 ///
311 /// // Register our socket.
312 /// os_queue.register(&mut socket, event::Id(0), Interests::READABLE, RegisterOption::EDGE)?;
313 ///
314 /// // Poll until our socket is ready.
315 /// while events.is_empty() {
316 /// poll::<_, io::Error>(&mut [&mut os_queue], &mut events, None)?;
317 /// }
318 ///
319 /// let mut buf = [0; 20];
320 /// let bytes_received = socket.recv(&mut buf)?;
321 /// println!("Received {:?} ({} bytes)", &buf[..bytes_received], bytes_received);
322 /// # assert_eq!(&buf[..bytes_received], b"Hello world");
323 /// #
324 /// # Ok(())
325 /// # }
326 /// ```
327 ///
328 /// # Notes
329 ///
330 /// This requires the socket to be [connected].
331 ///
332 /// [connected]: UdpSocket::connect
333 pub fn recv(&mut self, buf: &mut [u8]) -> io::Result<usize> {
334 self.socket.recv(buf)
335 }
336
337 /// Receives data from the socket, without removing it from the input queue.
338 /// On success, returns the number of bytes read and the address from whence
339 /// the data came.
340 ///
341 /// # Examples
342 ///
343 /// ```
344 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
345 /// use std::io;
346 ///
347 /// use gaea::net::UdpSocket;
348 /// use gaea::os::{RegisterOption, Interests};
349 /// use gaea::{event, OsQueue, poll};
350 ///
351 /// let mut os_queue = OsQueue::new()?;
352 /// let mut events = Vec::new();
353 ///
354 /// let address = "127.0.0.1:7011".parse()?;
355 /// let mut socket = UdpSocket::bind(address)?;
356 /// #
357 /// # // Send some data that we can receive.
358 /// # let mut socket2 = UdpSocket::bind("127.0.0.1:7111".parse()?)?;
359 /// # os_queue.register(&mut socket2, event::Id(1), Interests::WRITABLE, RegisterOption::EDGE)?;
360 /// # while events.is_empty() { poll::<_, io::Error>(&mut [&mut os_queue], &mut events, None)?; }
361 /// # let bytes_sent = socket2.send_to(b"Hello world", address)?;
362 /// # assert_eq!(bytes_sent, 11);
363 /// # events.clear();
364 ///
365 /// // Register our socket.
366 /// os_queue.register(&mut socket, event::Id(0), Interests::READABLE, RegisterOption::EDGE)?;
367 ///
368 /// // Poll until our socket is ready.
369 /// while events.is_empty() {
370 /// poll::<_, io::Error>(&mut [&mut os_queue], &mut events, None)?;
371 /// }
372 ///
373 /// let mut buf1 = [0; 20];
374 /// let (bytes_received1, from_address1) = socket.peek_from(&mut buf1)?;
375 /// println!("Peeked {:?} ({} bytes) from {}", &buf1[..bytes_received1], bytes_received1, from_address1);
376 /// # assert_eq!(&buf1[..bytes_received1], b"Hello world");
377 /// # assert_eq!(from_address1, "127.0.0.1:7111".parse()?);
378 ///
379 /// let mut buf2 = [0; 20];
380 /// let (bytes_received2, from_address2) = socket.recv_from(&mut buf2)?;
381 /// println!("Received {:?} ({} bytes) from {}", &buf2[..bytes_received2], bytes_received2, from_address2);
382 /// assert_eq!(bytes_received1, bytes_received2);
383 /// assert_eq!(&buf1[..bytes_received1], &buf2[..bytes_received2]);
384 /// assert_eq!(from_address1, from_address2);
385 /// #
386 /// # Ok(())
387 /// # }
388 /// ```
389 pub fn peek_from(&mut self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
390 self.socket.peek_from(buf)
391 }
392
393 /// Receives data from the socket, without removing it from the input queue.
394 /// On success, returns the number of bytes read.
395 ///
396 /// # Examples
397 ///
398 /// ```
399 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
400 /// use std::io;
401 ///
402 /// use gaea::net::UdpSocket;
403 /// use gaea::os::{RegisterOption, Interests};
404 /// use gaea::{event, OsQueue, poll};
405 ///
406 /// let mut os_queue = OsQueue::new()?;
407 /// let mut events = Vec::new();
408 ///
409 /// let local_address = "127.0.0.1:7012".parse()?;
410 /// let remote_address = "127.0.0.1:7013".parse()?;
411 /// let mut socket = UdpSocket::bind(local_address)?;
412 /// socket.connect(remote_address)?;
413 /// #
414 /// # // Send some data that we can receive.
415 /// # let mut socket2 = UdpSocket::bind(remote_address)?;
416 /// # os_queue.register(&mut socket2, event::Id(1), Interests::WRITABLE, RegisterOption::EDGE)?;
417 /// # while events.is_empty() { poll::<_, io::Error>(&mut [&mut os_queue], &mut events, None)?; }
418 /// # let bytes_sent = socket2.send_to(b"Hello world", local_address)?;
419 /// # assert_eq!(bytes_sent, 11);
420 /// # events.clear();
421 ///
422 /// // Register our socket.
423 /// os_queue.register(&mut socket, event::Id(0), Interests::READABLE, RegisterOption::EDGE)?;
424 ///
425 /// // Poll until our socket is ready.
426 /// while events.is_empty() {
427 /// poll::<_, io::Error>(&mut [&mut os_queue], &mut events, None)?;
428 /// }
429 ///
430 /// let mut buf1 = [0; 20];
431 /// let bytes_received1 = socket.peek(&mut buf1)?;
432 /// println!("Peeked {:?} ({} bytes)", &buf1[..bytes_received1], bytes_received1);
433 /// # assert_eq!(&buf1[..bytes_received1], b"Hello world");
434 ///
435 /// let mut buf2 = [0; 20];
436 /// let bytes_received2 = socket.recv(&mut buf2)?;
437 /// println!("Received {:?} ({} bytes)", &buf2[..bytes_received2], bytes_received2);
438 /// assert_eq!(bytes_received1, bytes_received2);
439 /// assert_eq!(&buf1[..bytes_received1], &buf2[..bytes_received2]);
440 /// #
441 /// # Ok(())
442 /// # }
443 /// ```
444 ///
445 /// # Notes
446 ///
447 /// This requires the socket to be [connected].
448 ///
449 /// [connected]: UdpSocket::connect
450 pub fn peek(&mut self, buf: &mut [u8]) -> io::Result<usize> {
451 self.socket.peek(buf)
452 }
453
454 /// Get the value of the `SO_ERROR` option on this socket.
455 ///
456 /// This will retrieve the stored error in the underlying socket, clearing
457 /// the field in the process. This can be useful for checking errors between
458 /// calls.
459 pub fn take_error(&mut self) -> io::Result<Option<io::Error>> {
460 self.socket.take_error()
461 }
462}
463
464impl Evented for UdpSocket {
465 fn register(&mut self, os_queue: &mut OsQueue, id: event::Id, interests: Interests, opt: RegisterOption) -> io::Result<()> {
466 self.socket.register(os_queue, id, interests, opt)
467 }
468
469 fn reregister(&mut self, os_queue: &mut OsQueue, id: event::Id, interests: Interests, opt: RegisterOption) -> io::Result<()> {
470 self.socket.reregister(os_queue, id, interests, opt)
471 }
472
473 fn deregister(&mut self, os_queue: &mut OsQueue) -> io::Result<()> {
474 self.socket.deregister(os_queue)
475 }
476}
477
478#[cfg(unix)]
479impl IntoRawFd for UdpSocket {
480 fn into_raw_fd(self) -> RawFd {
481 self.socket.into_raw_fd()
482 }
483}
484
485#[cfg(unix)]
486impl AsRawFd for UdpSocket {
487 fn as_raw_fd(&self) -> RawFd {
488 self.socket.as_raw_fd()
489 }
490}
491
492#[cfg(unix)]
493impl FromRawFd for UdpSocket {
494 unsafe fn from_raw_fd(fd: RawFd) -> UdpSocket {
495 UdpSocket {
496 socket: FromRawFd::from_raw_fd(fd),
497 }
498 }
499}