gio/socket.rs
1// Take a look at the license at the top of the repository in the LICENSE file.
2
3#[cfg(unix)]
4use std::os::unix::io::{AsFd, AsRawFd, BorrowedFd, FromRawFd, IntoRawFd, OwnedFd, RawFd};
5#[cfg(windows)]
6use std::os::windows::io::{
7 AsRawSocket, AsSocket, BorrowedSocket, FromRawSocket, IntoRawSocket, OwnedSocket, RawSocket,
8};
9#[cfg(feature = "v2_60")]
10use std::time::Duration;
11use std::{cell::RefCell, marker::PhantomData, mem::transmute, pin::Pin, ptr};
12
13use futures_core::stream::Stream;
14use glib::{Slice, prelude::*, translate::*};
15
16#[cfg(feature = "v2_60")]
17use crate::PollableReturn;
18use crate::{Cancellable, Socket, SocketAddress, SocketControlMessage, ffi};
19
20impl Socket {
21 /// Creates a new #GSocket from a native file descriptor
22 /// or winsock SOCKET handle.
23 ///
24 /// This reads all the settings from the file descriptor so that
25 /// all properties should work. Note that the file descriptor
26 /// will be set to non-blocking mode, independent on the blocking
27 /// mode of the #GSocket.
28 ///
29 /// On success, the returned #GSocket takes ownership of @fd. On failure, the
30 /// caller must close @fd themselves.
31 ///
32 /// Since GLib 2.46, it is no longer a fatal error to call this on a non-socket
33 /// descriptor. Instead, a GError will be set with code [`IOErrorEnum::Failed`][crate::IOErrorEnum::Failed]
34 /// ## `fd`
35 /// a native socket file descriptor.
36 ///
37 /// # Returns
38 ///
39 /// a #GSocket or [`None`] on error.
40 /// Free the returned object with g_object_unref().
41 #[cfg(unix)]
42 #[cfg_attr(docsrs, doc(cfg(unix)))]
43 #[doc(alias = "g_socket_new_from_fd")]
44 pub fn from_fd(fd: OwnedFd) -> Result<Socket, glib::Error> {
45 unsafe { Self::from_raw_fd(fd) }
46 }
47
48 #[cfg(unix)]
49 #[cfg_attr(docsrs, doc(cfg(unix)))]
50 #[doc(alias = "g_socket_new_from_fd")]
51 pub unsafe fn from_raw_fd(fd: impl IntoRawFd) -> Result<Socket, glib::Error> {
52 let fd = fd.into_raw_fd();
53 let mut error = ptr::null_mut();
54 unsafe {
55 let ret = ffi::g_socket_new_from_fd(fd, &mut error);
56 if error.is_null() {
57 Ok(from_glib_full(ret))
58 } else {
59 let _ = OwnedFd::from_raw_fd(fd);
60 Err(from_glib_full(error))
61 }
62 }
63 }
64
65 #[cfg(windows)]
66 #[cfg_attr(docsrs, doc(cfg(windows)))]
67 pub fn from_socket(socket: OwnedSocket) -> Result<Socket, glib::Error> {
68 unsafe { Self::from_raw_socket(socket) }
69 }
70
71 #[cfg(windows)]
72 #[cfg_attr(docsrs, doc(cfg(windows)))]
73 pub unsafe fn from_raw_socket(socket: impl IntoRawSocket) -> Result<Socket, glib::Error> {
74 let socket = socket.into_raw_socket();
75 let mut error = ptr::null_mut();
76 unsafe {
77 let ret = ffi::g_socket_new_from_fd(socket as i32, &mut error);
78 if error.is_null() {
79 Ok(from_glib_full(ret))
80 } else {
81 let _ = OwnedSocket::from_raw_socket(socket);
82 Err(from_glib_full(error))
83 }
84 }
85 }
86}
87
88#[cfg(unix)]
89#[cfg_attr(docsrs, doc(cfg(unix)))]
90impl AsRawFd for Socket {
91 fn as_raw_fd(&self) -> RawFd {
92 unsafe { ffi::g_socket_get_fd(self.to_glib_none().0) as _ }
93 }
94}
95
96#[cfg(unix)]
97#[cfg_attr(docsrs, doc(cfg(unix)))]
98impl AsFd for Socket {
99 fn as_fd(&self) -> BorrowedFd<'_> {
100 unsafe {
101 let raw_fd = self.as_raw_fd();
102 BorrowedFd::borrow_raw(raw_fd)
103 }
104 }
105}
106
107#[cfg(windows)]
108#[cfg_attr(docsrs, doc(cfg(windows)))]
109impl AsRawSocket for Socket {
110 fn as_raw_socket(&self) -> RawSocket {
111 unsafe { ffi::g_socket_get_fd(self.to_glib_none().0) as _ }
112 }
113}
114
115#[cfg(windows)]
116#[cfg_attr(docsrs, doc(cfg(windows)))]
117impl AsSocket for Socket {
118 fn as_socket(&self) -> BorrowedSocket<'_> {
119 unsafe {
120 let raw_socket = self.as_raw_socket();
121 BorrowedSocket::borrow_raw(raw_socket)
122 }
123 }
124}
125
126#[doc(alias = "GInputVector")]
127#[repr(transparent)]
128#[derive(Debug)]
129pub struct InputVector<'v> {
130 vector: ffi::GInputVector,
131 buffer: PhantomData<&'v mut [u8]>,
132}
133
134impl<'v> InputVector<'v> {
135 #[inline]
136 pub fn new(buffer: &'v mut [u8]) -> Self {
137 Self {
138 vector: ffi::GInputVector {
139 buffer: buffer.as_mut_ptr() as *mut _,
140 size: buffer.len(),
141 },
142 buffer: PhantomData,
143 }
144 }
145}
146
147unsafe impl Send for InputVector<'_> {}
148unsafe impl Sync for InputVector<'_> {}
149
150impl std::ops::Deref for InputVector<'_> {
151 type Target = [u8];
152
153 #[inline]
154 fn deref(&self) -> &Self::Target {
155 unsafe { std::slice::from_raw_parts(self.vector.buffer as *const _, self.vector.size) }
156 }
157}
158
159impl std::ops::DerefMut for InputVector<'_> {
160 #[inline]
161 fn deref_mut(&mut self) -> &mut Self::Target {
162 unsafe { std::slice::from_raw_parts_mut(self.vector.buffer as *mut _, self.vector.size) }
163 }
164}
165
166#[derive(Debug)]
167pub struct SocketControlMessages {
168 ptr: *mut *mut ffi::GSocketControlMessage,
169 len: u32,
170}
171
172impl SocketControlMessages {
173 #[inline]
174 pub const fn new() -> Self {
175 Self {
176 ptr: ptr::null_mut(),
177 len: 0,
178 }
179 }
180}
181
182impl AsRef<[SocketControlMessage]> for SocketControlMessages {
183 #[inline]
184 fn as_ref(&self) -> &[SocketControlMessage] {
185 unsafe { std::slice::from_raw_parts(self.ptr as *const _, self.len as usize) }
186 }
187}
188
189impl std::ops::Deref for SocketControlMessages {
190 type Target = [SocketControlMessage];
191
192 #[inline]
193 fn deref(&self) -> &Self::Target {
194 self.as_ref()
195 }
196}
197
198impl Default for SocketControlMessages {
199 #[inline]
200 fn default() -> Self {
201 Self::new()
202 }
203}
204
205impl Drop for SocketControlMessages {
206 #[inline]
207 fn drop(&mut self) {
208 unsafe {
209 let _: Slice<SocketControlMessage> =
210 Slice::from_glib_full_num(self.ptr as *mut _, self.len as usize);
211 }
212 }
213}
214
215/// Structure used for scatter/gather data input when receiving multiple
216/// messages or packets in one go. You generally pass in an array of empty
217/// #GInputVectors and the operation will use all the buffers as if they
218/// were one buffer, and will set @bytes_received to the total number of bytes
219/// received across all #GInputVectors.
220///
221/// This structure closely mirrors `struct mmsghdr` and `struct msghdr` from
222/// the POSIX sockets API (see `man 2 recvmmsg`).
223///
224/// If @address is non-[`None`] then it is set to the source address the message
225/// was received from (or [`None`] if it is not available), and the caller must
226/// free it afterwards.
227///
228/// If @control_messages is non-[`None`] then it is set to an array of control
229/// messages received with the message (if any), and the caller must free it
230/// afterwards. @num_control_messages is set to the number of elements in
231/// this array, which may be zero.
232///
233/// Flags relevant to this message will be returned in @flags. For example,
234/// `MSG_EOR` or `MSG_TRUNC`.
235#[doc(alias = "GInputMessage")]
236#[repr(transparent)]
237#[derive(Debug)]
238pub struct InputMessage<'m> {
239 message: ffi::GInputMessage,
240 address: PhantomData<Option<&'m mut Option<SocketAddress>>>,
241 vectors: PhantomData<&'m mut [InputVector<'m>]>,
242 control_messages: PhantomData<Option<&'m mut SocketControlMessages>>,
243}
244
245impl<'m> InputMessage<'m> {
246 pub fn new(
247 mut address: Option<&'m mut Option<SocketAddress>>,
248 vectors: &'m mut [InputVector<'m>],
249 control_messages: Option<&'m mut SocketControlMessages>,
250 ) -> Self {
251 let address = address
252 .as_mut()
253 .map(|a| {
254 assert!(a.is_none());
255 *a as *mut _ as *mut _
256 })
257 .unwrap_or_else(ptr::null_mut);
258 let (control_messages, num_control_messages) = control_messages
259 .map(|c| (&mut c.ptr as *mut _, &mut c.len as *mut _))
260 .unwrap_or_else(|| (ptr::null_mut(), ptr::null_mut()));
261 Self {
262 message: ffi::GInputMessage {
263 address,
264 vectors: vectors.as_mut_ptr() as *mut ffi::GInputVector,
265 num_vectors: vectors.len().try_into().unwrap(),
266 bytes_received: 0,
267 flags: 0,
268 control_messages,
269 num_control_messages,
270 },
271 address: PhantomData,
272 vectors: PhantomData,
273 control_messages: PhantomData,
274 }
275 }
276 #[inline]
277 pub fn vectors(&mut self) -> &'m mut [InputVector<'m>] {
278 unsafe {
279 std::slice::from_raw_parts_mut(
280 self.message.vectors as *mut _,
281 self.message.num_vectors as usize,
282 )
283 }
284 }
285 #[inline]
286 pub const fn flags(&self) -> i32 {
287 self.message.flags
288 }
289 #[inline]
290 pub const fn bytes_received(&self) -> usize {
291 self.message.bytes_received
292 }
293}
294
295#[doc(alias = "GOutputVector")]
296#[repr(transparent)]
297#[derive(Debug)]
298pub struct OutputVector<'v> {
299 vector: ffi::GOutputVector,
300 buffer: PhantomData<&'v [u8]>,
301}
302
303impl<'v> OutputVector<'v> {
304 #[inline]
305 pub const fn new(buffer: &'v [u8]) -> Self {
306 Self {
307 vector: ffi::GOutputVector {
308 buffer: buffer.as_ptr() as *const _,
309 size: buffer.len(),
310 },
311 buffer: PhantomData,
312 }
313 }
314}
315
316unsafe impl Send for OutputVector<'_> {}
317unsafe impl Sync for OutputVector<'_> {}
318
319impl std::ops::Deref for OutputVector<'_> {
320 type Target = [u8];
321
322 #[inline]
323 fn deref(&self) -> &Self::Target {
324 unsafe { std::slice::from_raw_parts(self.vector.buffer as *const _, self.vector.size) }
325 }
326}
327
328/// Structure used for scatter/gather data output when sending multiple
329/// messages or packets in one go. You generally pass in an array of
330/// #GOutputVectors and the operation will use all the buffers as if they
331/// were one buffer.
332///
333/// If @address is [`None`] then the message is sent to the default receiver
334/// (as previously set by g_socket_connect()).
335#[doc(alias = "GOutputMessage")]
336#[repr(transparent)]
337#[derive(Debug)]
338pub struct OutputMessage<'m> {
339 message: ffi::GOutputMessage,
340 address: PhantomData<Option<&'m SocketAddress>>,
341 vectors: PhantomData<&'m [OutputVector<'m>]>,
342 control_messages: PhantomData<&'m [SocketControlMessage]>,
343}
344
345impl<'m> OutputMessage<'m> {
346 pub fn new<A: IsA<SocketAddress>>(
347 address: Option<&'m A>,
348 vectors: &'m [OutputVector<'m>],
349 control_messages: &'m [SocketControlMessage],
350 ) -> Self {
351 Self {
352 message: ffi::GOutputMessage {
353 address: address
354 .map(|a| a.upcast_ref::<SocketAddress>().as_ptr())
355 .unwrap_or_else(ptr::null_mut),
356 vectors: mut_override(vectors.as_ptr() as *const ffi::GOutputVector),
357 num_vectors: vectors.len().try_into().unwrap(),
358 bytes_sent: 0,
359 control_messages: control_messages.as_ptr() as *mut _,
360 num_control_messages: control_messages.len().try_into().unwrap(),
361 },
362 address: PhantomData,
363 vectors: PhantomData,
364 control_messages: PhantomData,
365 }
366 }
367 #[inline]
368 pub fn vectors(&self) -> &'m [OutputVector<'m>] {
369 unsafe {
370 std::slice::from_raw_parts(
371 self.message.vectors as *const _,
372 self.message.num_vectors as usize,
373 )
374 }
375 }
376 #[inline]
377 pub fn bytes_sent(&self) -> u32 {
378 self.message.bytes_sent
379 }
380}
381
382pub trait SocketExtManual: IsA<Socket> + Sized {
383 /// Receive data (up to @size bytes) from a socket. This is mainly used by
384 /// connection-oriented sockets; it is identical to g_socket_receive_from()
385 /// with @address set to [`None`].
386 ///
387 /// For [`SocketType::Datagram`][crate::SocketType::Datagram] and [`SocketType::Seqpacket`][crate::SocketType::Seqpacket] sockets,
388 /// g_socket_receive() will always read either 0 or 1 complete messages from
389 /// the socket. If the received message is too large to fit in @buffer, then
390 /// the data beyond @size bytes will be discarded, without any explicit
391 /// indication that this has occurred.
392 ///
393 /// For [`SocketType::Stream`][crate::SocketType::Stream] sockets, g_socket_receive() can return any
394 /// number of bytes, up to @size. If more than @size bytes have been
395 /// received, the additional data will be returned in future calls to
396 /// g_socket_receive().
397 ///
398 /// If the socket is in blocking mode the call will block until there
399 /// is some data to receive, the connection is closed, or there is an
400 /// error. If there is no data available and the socket is in
401 /// non-blocking mode, a [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] error will be
402 /// returned. To be notified when data is available, wait for the
403 /// [`glib::IOCondition::IN`][crate::glib::IOCondition::IN] condition.
404 ///
405 /// On error -1 is returned and @error is set accordingly.
406 /// ## `cancellable`
407 /// a `GCancellable` or [`None`]
408 ///
409 /// # Returns
410 ///
411 /// Number of bytes read, or 0 if the connection was closed by
412 /// the peer, or -1 on error
413 ///
414 /// ## `buffer`
415 ///
416 /// a buffer to read data into (which should be at least @size bytes long).
417 #[doc(alias = "g_socket_receive")]
418 fn receive<B: AsMut<[u8]>, C: IsA<Cancellable>>(
419 &self,
420 mut buffer: B,
421 cancellable: Option<&C>,
422 ) -> Result<usize, glib::Error> {
423 let cancellable = cancellable.map(|c| c.as_ref());
424 let gcancellable = cancellable.to_glib_none();
425 let buffer = buffer.as_mut();
426 let buffer_ptr = buffer.as_mut_ptr();
427 let count = buffer.len();
428 unsafe {
429 let mut error = ptr::null_mut();
430 let ret = ffi::g_socket_receive(
431 self.as_ref().to_glib_none().0,
432 buffer_ptr,
433 count,
434 gcancellable.0,
435 &mut error,
436 );
437 if error.is_null() {
438 Ok(ret as usize)
439 } else {
440 Err(from_glib_full(error))
441 }
442 }
443 }
444 /// Receive data (up to @size bytes) from a socket.
445 ///
446 /// If @address is non-[`None`] then @address will be set equal to the
447 /// source address of the received packet, or [`None`] if it is not available.
448 /// @address is owned by the caller.
449 ///
450 /// See g_socket_receive() for additional information.
451 /// ## `cancellable`
452 /// a `GCancellable` or [`None`]
453 ///
454 /// # Returns
455 ///
456 /// Number of bytes read, or 0 if the connection was closed by
457 /// the peer, or -1 on error
458 ///
459 /// ## `address`
460 /// a pointer to a #GSocketAddress
461 /// pointer, or [`None`]
462 ///
463 /// ## `buffer`
464 ///
465 /// a buffer to read data into (which should be at least @size bytes long).
466 #[doc(alias = "g_socket_receive_from")]
467 fn receive_from<B: AsMut<[u8]>, C: IsA<Cancellable>>(
468 &self,
469 mut buffer: B,
470 cancellable: Option<&C>,
471 ) -> Result<(usize, Option<SocketAddress>), glib::Error> {
472 let cancellable = cancellable.map(|c| c.as_ref());
473 let gcancellable = cancellable.to_glib_none();
474 let buffer = buffer.as_mut();
475 let buffer_ptr = buffer.as_mut_ptr();
476 let count = buffer.len();
477 unsafe {
478 let mut error = ptr::null_mut();
479 let mut addr_ptr = ptr::null_mut();
480
481 let ret = ffi::g_socket_receive_from(
482 self.as_ref().to_glib_none().0,
483 &mut addr_ptr,
484 buffer_ptr,
485 count,
486 gcancellable.0,
487 &mut error,
488 );
489 if error.is_null() {
490 Ok((ret as usize, from_glib_full(addr_ptr)))
491 } else {
492 Err(from_glib_full(error))
493 }
494 }
495 }
496 /// Receive data from a socket. For receiving multiple messages, see
497 /// g_socket_receive_messages(); for easier use, see
498 /// g_socket_receive() and g_socket_receive_from().
499 ///
500 /// If @address is non-[`None`] then @address will be set equal to the
501 /// source address of the received packet, or [`None`] if it is not available.
502 /// @address is owned by the caller.
503 ///
504 /// @vector must point to an array of #GInputVector structs and
505 /// @num_vectors must be the length of this array. These structs
506 /// describe the buffers that received data will be scattered into.
507 /// If @num_vectors is -1, then @vectors is assumed to be terminated
508 /// by a #GInputVector with a [`None`] buffer pointer.
509 ///
510 /// As a special case, if @num_vectors is 0 (in which case, @vectors
511 /// may of course be [`None`]), then a single byte is received and
512 /// discarded. This is to facilitate the common practice of sending a
513 /// single '\0' byte for the purposes of transferring ancillary data.
514 ///
515 /// @messages, if non-[`None`], will be set to point to a newly-allocated
516 /// array of #GSocketControlMessage instances or [`None`] if no such
517 /// messages was received. These correspond to the control messages
518 /// received from the kernel, one #GSocketControlMessage per message
519 /// from the kernel. This array is [`None`]-terminated and must be freed
520 /// by the caller using g_free() after calling g_object_unref() on each
521 /// element. If @messages is [`None`], any control messages received will
522 /// be discarded.
523 ///
524 /// @num_messages, if non-[`None`], will be set to the number of control
525 /// messages received.
526 ///
527 /// If both @messages and @num_messages are non-[`None`], then
528 /// @num_messages gives the number of #GSocketControlMessage instances
529 /// in @messages (ie: not including the [`None`] terminator).
530 ///
531 /// @flags is an in/out parameter. The commonly available arguments
532 /// for this are available in the #GSocketMsgFlags enum, but the
533 /// values there are the same as the system values, and the flags
534 /// are passed in as-is, so you can pass in system-specific flags too
535 /// (and g_socket_receive_message() may pass system-specific flags out).
536 /// Flags passed in to the parameter affect the receive operation; flags returned
537 /// out of it are relevant to the specific returned message.
538 ///
539 /// As with g_socket_receive(), data may be discarded if @self is
540 /// [`SocketType::Datagram`][crate::SocketType::Datagram] or [`SocketType::Seqpacket`][crate::SocketType::Seqpacket] and you do not
541 /// provide enough buffer space to read a complete message. You can pass
542 /// [`SocketMsgFlags::PEEK`][crate::SocketMsgFlags::PEEK] in @flags to peek at the current message without
543 /// removing it from the receive queue, but there is no portable way to find
544 /// out the length of the message other than by reading it into a
545 /// sufficiently-large buffer.
546 ///
547 /// If the socket is in blocking mode the call will block until there
548 /// is some data to receive, the connection is closed, or there is an
549 /// error. If there is no data available and the socket is in
550 /// non-blocking mode, a [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] error will be
551 /// returned. To be notified when data is available, wait for the
552 /// [`glib::IOCondition::IN`][crate::glib::IOCondition::IN] condition.
553 ///
554 /// On error -1 is returned and @error is set accordingly.
555 /// ## `vectors`
556 /// an array of #GInputVector structs
557 /// ## `cancellable`
558 /// a `GCancellable` or [`None`]
559 ///
560 /// # Returns
561 ///
562 /// Number of bytes read, or 0 if the connection was closed by
563 /// the peer, or -1 on error
564 ///
565 /// ## `address`
566 /// a pointer to a #GSocketAddress
567 /// pointer, or [`None`]
568 ///
569 /// ## `messages`
570 /// a pointer
571 /// which may be filled with an array of #GSocketControlMessages, or [`None`]
572 ///
573 /// ## `flags`
574 /// a pointer to an int containing #GSocketMsgFlags flags,
575 /// which may additionally contain
576 /// [other platform specific flags](http://man7.org/linux/man-pages/man2/recv.2.html)
577 #[doc(alias = "g_socket_receive_message")]
578 fn receive_message<C: IsA<Cancellable>>(
579 &self,
580 mut address: Option<&mut Option<SocketAddress>>,
581 vectors: &mut [InputVector],
582 control_messages: Option<&mut SocketControlMessages>,
583 mut flags: i32,
584 cancellable: Option<&C>,
585 ) -> Result<(usize, i32), glib::Error> {
586 let cancellable = cancellable.map(|c| c.as_ref());
587 let address = address
588 .as_mut()
589 .map(|a| {
590 assert!(a.is_none());
591 *a as *mut _ as *mut _
592 })
593 .unwrap_or_else(ptr::null_mut);
594 let (control_messages, num_control_messages) = control_messages
595 .map(|c| (&mut c.ptr as *mut _, &mut c.len as *mut _ as *mut _))
596 .unwrap_or_else(|| (ptr::null_mut(), ptr::null_mut()));
597 unsafe {
598 let mut error = ptr::null_mut();
599
600 let received = ffi::g_socket_receive_message(
601 self.as_ref().to_glib_none().0,
602 address,
603 vectors.as_mut_ptr() as *mut ffi::GInputVector,
604 vectors.len().try_into().unwrap(),
605 control_messages,
606 num_control_messages,
607 &mut flags,
608 cancellable.to_glib_none().0,
609 &mut error,
610 );
611 if error.is_null() {
612 Ok((received as usize, flags))
613 } else {
614 Err(from_glib_full(error))
615 }
616 }
617 }
618 /// Receive multiple data messages from @self in one go. This is the most
619 /// complicated and fully-featured version of this call. For easier use, see
620 /// g_socket_receive(), g_socket_receive_from(), and g_socket_receive_message().
621 ///
622 /// @messages must point to an array of #GInputMessage structs and
623 /// @num_messages must be the length of this array. Each #GInputMessage
624 /// contains a pointer to an array of #GInputVector structs describing the
625 /// buffers that the data received in each message will be written to. Using
626 /// multiple #GInputVectors is more memory-efficient than manually copying data
627 /// out of a single buffer to multiple sources, and more system-call-efficient
628 /// than making multiple calls to g_socket_receive(), such as in scenarios where
629 /// a lot of data packets need to be received (e.g. high-bandwidth video
630 /// streaming over RTP/UDP).
631 ///
632 /// @flags modify how all messages are received. The commonly available
633 /// arguments for this are available in the #GSocketMsgFlags enum, but the
634 /// values there are the same as the system values, and the flags
635 /// are passed in as-is, so you can pass in system-specific flags too. These
636 /// flags affect the overall receive operation. Flags affecting individual
637 /// messages are returned in #GInputMessage.flags.
638 ///
639 /// The other members of #GInputMessage are treated as described in its
640 /// documentation.
641 ///
642 /// If #GSocket:blocking is [`true`] the call will block until @num_messages have
643 /// been received, or the end of the stream is reached.
644 ///
645 /// If #GSocket:blocking is [`false`] the call will return up to @num_messages
646 /// without blocking, or [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] if no messages are queued in the
647 /// operating system to be received.
648 ///
649 /// In blocking mode, if #GSocket:timeout is positive and is reached before any
650 /// messages are received, [`IOErrorEnum::TimedOut`][crate::IOErrorEnum::TimedOut] is returned, otherwise up to
651 /// @num_messages are returned. (Note: This is effectively the
652 /// behaviour of `MSG_WAITFORONE` with recvmmsg().)
653 ///
654 /// To be notified when messages are available, wait for the
655 /// [`glib::IOCondition::IN`][crate::glib::IOCondition::IN] condition. Note though that you may still receive
656 /// [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] from g_socket_receive_messages() even if you were
657 /// previously notified of a [`glib::IOCondition::IN`][crate::glib::IOCondition::IN] condition.
658 ///
659 /// If the remote peer closes the connection, any messages queued in the
660 /// operating system will be returned, and subsequent calls to
661 /// g_socket_receive_messages() will return 0 (with no error set).
662 ///
663 /// On error -1 is returned and @error is set accordingly. An error will only
664 /// be returned if zero messages could be received; otherwise the number of
665 /// messages successfully received before the error will be returned.
666 /// ## `messages`
667 /// an array of #GInputMessage structs
668 /// ## `flags`
669 /// an int containing #GSocketMsgFlags flags for the overall operation,
670 /// which may additionally contain
671 /// [other platform specific flags](http://man7.org/linux/man-pages/man2/recv.2.html)
672 /// ## `cancellable`
673 /// a `GCancellable` or [`None`]
674 ///
675 /// # Returns
676 ///
677 /// number of messages received, or -1 on error. Note that the number
678 /// of messages received may be smaller than @num_messages if in non-blocking
679 /// mode, if the peer closed the connection, or if @num_messages
680 /// was larger than `UIO_MAXIOV` (1024), in which case the caller may re-try
681 /// to receive the remaining messages.
682 #[doc(alias = "g_socket_receive_messages")]
683 fn receive_messages<C: IsA<Cancellable>>(
684 &self,
685 messages: &mut [InputMessage],
686 flags: i32,
687 cancellable: Option<&C>,
688 ) -> Result<usize, glib::Error> {
689 let cancellable = cancellable.map(|c| c.as_ref());
690 unsafe {
691 let mut error = ptr::null_mut();
692
693 let count = ffi::g_socket_receive_messages(
694 self.as_ref().to_glib_none().0,
695 messages.as_mut_ptr() as *mut _,
696 messages.len().try_into().unwrap(),
697 flags,
698 cancellable.to_glib_none().0,
699 &mut error,
700 );
701 if error.is_null() {
702 Ok(count as usize)
703 } else {
704 Err(from_glib_full(error))
705 }
706 }
707 }
708 /// This behaves exactly the same as g_socket_receive(), except that
709 /// the choice of blocking or non-blocking behavior is determined by
710 /// the @blocking argument rather than by @self's properties.
711 /// ## `blocking`
712 /// whether to do blocking or non-blocking I/O
713 /// ## `cancellable`
714 /// a `GCancellable` or [`None`]
715 ///
716 /// # Returns
717 ///
718 /// Number of bytes read, or 0 if the connection was closed by
719 /// the peer, or -1 on error
720 ///
721 /// ## `buffer`
722 ///
723 /// a buffer to read data into (which should be at least @size bytes long).
724 #[doc(alias = "g_socket_receive_with_blocking")]
725 fn receive_with_blocking<B: AsMut<[u8]>, C: IsA<Cancellable>>(
726 &self,
727 mut buffer: B,
728 blocking: bool,
729 cancellable: Option<&C>,
730 ) -> Result<usize, glib::Error> {
731 let cancellable = cancellable.map(|c| c.as_ref());
732 let gcancellable = cancellable.to_glib_none();
733 let buffer = buffer.as_mut();
734 let buffer_ptr = buffer.as_mut_ptr();
735 let count = buffer.len();
736 unsafe {
737 let mut error = ptr::null_mut();
738 let ret = ffi::g_socket_receive_with_blocking(
739 self.as_ref().to_glib_none().0,
740 buffer_ptr,
741 count,
742 blocking.into_glib(),
743 gcancellable.0,
744 &mut error,
745 );
746 if error.is_null() {
747 Ok(ret as usize)
748 } else {
749 Err(from_glib_full(error))
750 }
751 }
752 }
753
754 /// Tries to send @size bytes from @buffer on the socket. This is
755 /// mainly used by connection-oriented sockets; it is identical to
756 /// g_socket_send_to() with @address set to [`None`].
757 ///
758 /// If the socket is in blocking mode the call will block until there is
759 /// space for the data in the socket queue. If there is no space available
760 /// and the socket is in non-blocking mode a [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] error
761 /// will be returned. To be notified when space is available, wait for the
762 /// [`glib::IOCondition::OUT`][crate::glib::IOCondition::OUT] condition. Note though that you may still receive
763 /// [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] from g_socket_send() even if you were previously
764 /// notified of a [`glib::IOCondition::OUT`][crate::glib::IOCondition::OUT] condition. (On Windows in particular, this is
765 /// very common due to the way the underlying APIs work.)
766 ///
767 /// On error -1 is returned and @error is set accordingly.
768 /// ## `buffer`
769 /// the buffer
770 /// containing the data to send.
771 /// ## `cancellable`
772 /// a `GCancellable` or [`None`]
773 ///
774 /// # Returns
775 ///
776 /// Number of bytes written (which may be less than @size), or -1
777 /// on error
778 #[doc(alias = "g_socket_send")]
779 fn send<B: AsRef<[u8]>, C: IsA<Cancellable>>(
780 &self,
781 buffer: B,
782 cancellable: Option<&C>,
783 ) -> Result<usize, glib::Error> {
784 let cancellable = cancellable.map(|c| c.as_ref());
785 let gcancellable = cancellable.to_glib_none();
786 let (count, buffer_ptr) = {
787 let slice = buffer.as_ref();
788 (slice.len(), slice.as_ptr())
789 };
790 unsafe {
791 let mut error = ptr::null_mut();
792 let ret = ffi::g_socket_send(
793 self.as_ref().to_glib_none().0,
794 mut_override(buffer_ptr),
795 count,
796 gcancellable.0,
797 &mut error,
798 );
799 if error.is_null() {
800 Ok(ret as usize)
801 } else {
802 Err(from_glib_full(error))
803 }
804 }
805 }
806 /// Send data to @address on @self. For sending multiple messages see
807 /// g_socket_send_messages(); for easier use, see
808 /// g_socket_send() and g_socket_send_to().
809 ///
810 /// If @address is [`None`] then the message is sent to the default receiver
811 /// (set by g_socket_connect()).
812 ///
813 /// @vectors must point to an array of #GOutputVector structs and
814 /// @num_vectors must be the length of this array. (If @num_vectors is -1,
815 /// then @vectors is assumed to be terminated by a #GOutputVector with a
816 /// [`None`] buffer pointer.) The #GOutputVector structs describe the buffers
817 /// that the sent data will be gathered from. Using multiple
818 /// #GOutputVectors is more memory-efficient than manually copying
819 /// data from multiple sources into a single buffer, and more
820 /// network-efficient than making multiple calls to g_socket_send().
821 ///
822 /// @messages, if non-[`None`], is taken to point to an array of @num_messages
823 /// #GSocketControlMessage instances. These correspond to the control
824 /// messages to be sent on the socket.
825 /// If @num_messages is -1 then @messages is treated as a [`None`]-terminated
826 /// array.
827 ///
828 /// @flags modify how the message is sent. The commonly available arguments
829 /// for this are available in the #GSocketMsgFlags enum, but the
830 /// values there are the same as the system values, and the flags
831 /// are passed in as-is, so you can pass in system-specific flags too.
832 ///
833 /// If the socket is in blocking mode the call will block until there is
834 /// space for the data in the socket queue. If there is no space available
835 /// and the socket is in non-blocking mode a [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] error
836 /// will be returned. To be notified when space is available, wait for the
837 /// [`glib::IOCondition::OUT`][crate::glib::IOCondition::OUT] condition. Note though that you may still receive
838 /// [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] from g_socket_send() even if you were previously
839 /// notified of a [`glib::IOCondition::OUT`][crate::glib::IOCondition::OUT] condition. (On Windows in particular, this is
840 /// very common due to the way the underlying APIs work.)
841 ///
842 /// The sum of the sizes of each #GOutputVector in vectors must not be
843 /// greater than `G_MAXSSIZE`. If the message can be larger than this,
844 /// then it is mandatory to use the g_socket_send_message_with_timeout()
845 /// function.
846 ///
847 /// On error -1 is returned and @error is set accordingly.
848 /// ## `address`
849 /// a #GSocketAddress, or [`None`]
850 /// ## `vectors`
851 /// an array of #GOutputVector structs
852 /// ## `messages`
853 /// a pointer to an
854 /// array of #GSocketControlMessages, or [`None`].
855 /// ## `flags`
856 /// an int containing #GSocketMsgFlags flags, which may additionally
857 /// contain [other platform specific flags](http://man7.org/linux/man-pages/man2/recv.2.html)
858 /// ## `cancellable`
859 /// a `GCancellable` or [`None`]
860 ///
861 /// # Returns
862 ///
863 /// Number of bytes written (which may be less than @size), or -1
864 /// on error
865 #[doc(alias = "g_socket_send_message")]
866 fn send_message<P: IsA<SocketAddress>, C: IsA<Cancellable>>(
867 &self,
868 address: Option<&P>,
869 vectors: &[OutputVector],
870 messages: &[SocketControlMessage],
871 flags: i32,
872 cancellable: Option<&C>,
873 ) -> Result<usize, glib::Error> {
874 let cancellable = cancellable.map(|c| c.as_ref());
875 unsafe {
876 let mut error = ptr::null_mut();
877 let ret = ffi::g_socket_send_message(
878 self.as_ref().to_glib_none().0,
879 address.map(|p| p.as_ref()).to_glib_none().0,
880 vectors.as_ptr() as *mut ffi::GOutputVector,
881 vectors.len().try_into().unwrap(),
882 messages.as_ptr() as *mut _,
883 messages.len().try_into().unwrap(),
884 flags,
885 cancellable.to_glib_none().0,
886 &mut error,
887 );
888 if error.is_null() {
889 Ok(ret as usize)
890 } else {
891 Err(from_glib_full(error))
892 }
893 }
894 }
895 /// This behaves exactly the same as g_socket_send_message(), except that
896 /// the choice of timeout behavior is determined by the @timeout_us argument
897 /// rather than by @self's properties.
898 ///
899 /// On error [`PollableReturn::Failed`][crate::PollableReturn::Failed] is returned and @error is set accordingly, or
900 /// if the socket is currently not writable [`PollableReturn::WouldBlock`][crate::PollableReturn::WouldBlock] is
901 /// returned. @bytes_written will contain 0 in both cases.
902 /// ## `address`
903 /// a #GSocketAddress, or [`None`]
904 /// ## `vectors`
905 /// an array of #GOutputVector structs
906 /// ## `messages`
907 /// a pointer to an
908 /// array of #GSocketControlMessages, or [`None`].
909 /// ## `flags`
910 /// an int containing #GSocketMsgFlags flags, which may additionally
911 /// contain [other platform specific flags](http://man7.org/linux/man-pages/man2/recv.2.html)
912 /// ## `timeout_us`
913 /// the maximum time (in microseconds) to wait, or -1
914 /// ## `cancellable`
915 /// a `GCancellable` or [`None`]
916 ///
917 /// # Returns
918 ///
919 /// [`PollableReturn::Ok`][crate::PollableReturn::Ok] if all data was successfully written,
920 /// [`PollableReturn::WouldBlock`][crate::PollableReturn::WouldBlock] if the socket is currently not writable, or
921 /// [`PollableReturn::Failed`][crate::PollableReturn::Failed] if an error happened and @error is set.
922 ///
923 /// ## `bytes_written`
924 /// location to store the number of bytes that were written to the socket
925 #[cfg(feature = "v2_60")]
926 #[cfg_attr(docsrs, doc(cfg(feature = "v2_60")))]
927 #[doc(alias = "g_socket_send_message_with_timeout")]
928 fn send_message_with_timeout<P: IsA<SocketAddress>, C: IsA<Cancellable>>(
929 &self,
930 address: Option<&P>,
931 vectors: &[OutputVector],
932 messages: &[SocketControlMessage],
933 flags: i32,
934 timeout: Option<Duration>,
935 cancellable: Option<&C>,
936 ) -> Result<(PollableReturn, usize), glib::Error> {
937 let cancellable = cancellable.map(|c| c.as_ref());
938 unsafe {
939 let mut error = ptr::null_mut();
940 let mut bytes_written = 0;
941
942 let ret = ffi::g_socket_send_message_with_timeout(
943 self.as_ref().to_glib_none().0,
944 address.map(|p| p.as_ref()).to_glib_none().0,
945 vectors.as_ptr() as *mut ffi::GOutputVector,
946 vectors.len().try_into().unwrap(),
947 messages.as_ptr() as *mut _,
948 messages.len().try_into().unwrap(),
949 flags,
950 timeout
951 .map(|t| t.as_micros().try_into().unwrap())
952 .unwrap_or(-1),
953 &mut bytes_written,
954 cancellable.to_glib_none().0,
955 &mut error,
956 );
957 if error.is_null() {
958 Ok((from_glib(ret), bytes_written))
959 } else {
960 Err(from_glib_full(error))
961 }
962 }
963 }
964 /// Send multiple data messages from @self in one go. This is the most
965 /// complicated and fully-featured version of this call. For easier use, see
966 /// g_socket_send(), g_socket_send_to(), and g_socket_send_message().
967 ///
968 /// @messages must point to an array of #GOutputMessage structs and
969 /// @num_messages must be the length of this array. Each #GOutputMessage
970 /// contains an address to send the data to, and a pointer to an array of
971 /// #GOutputVector structs to describe the buffers that the data to be sent
972 /// for each message will be gathered from. Using multiple #GOutputVectors is
973 /// more memory-efficient than manually copying data from multiple sources
974 /// into a single buffer, and more network-efficient than making multiple
975 /// calls to g_socket_send(). Sending multiple messages in one go avoids the
976 /// overhead of making a lot of syscalls in scenarios where a lot of data
977 /// packets need to be sent (e.g. high-bandwidth video streaming over RTP/UDP),
978 /// or where the same data needs to be sent to multiple recipients.
979 ///
980 /// @flags modify how the message is sent. The commonly available arguments
981 /// for this are available in the #GSocketMsgFlags enum, but the
982 /// values there are the same as the system values, and the flags
983 /// are passed in as-is, so you can pass in system-specific flags too.
984 ///
985 /// If the socket is in blocking mode the call will block until there is
986 /// space for all the data in the socket queue. If there is no space available
987 /// and the socket is in non-blocking mode a [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] error
988 /// will be returned if no data was written at all, otherwise the number of
989 /// messages sent will be returned. To be notified when space is available,
990 /// wait for the [`glib::IOCondition::OUT`][crate::glib::IOCondition::OUT] condition. Note though that you may still receive
991 /// [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] from g_socket_send() even if you were previously
992 /// notified of a [`glib::IOCondition::OUT`][crate::glib::IOCondition::OUT] condition. (On Windows in particular, this is
993 /// very common due to the way the underlying APIs work.)
994 ///
995 /// On error -1 is returned and @error is set accordingly. An error will only
996 /// be returned if zero messages could be sent; otherwise the number of messages
997 /// successfully sent before the error will be returned.
998 /// ## `messages`
999 /// an array of #GOutputMessage structs
1000 /// ## `flags`
1001 /// an int containing #GSocketMsgFlags flags, which may additionally
1002 /// contain [other platform specific flags](http://man7.org/linux/man-pages/man2/recv.2.html)
1003 /// ## `cancellable`
1004 /// a `GCancellable` or [`None`]
1005 ///
1006 /// # Returns
1007 ///
1008 /// number of messages sent, or -1 on error. Note that the number of
1009 /// messages sent may be smaller than @num_messages if the socket is
1010 /// non-blocking or if @num_messages was larger than UIO_MAXIOV (1024),
1011 /// in which case the caller may re-try to send the remaining messages.
1012 #[doc(alias = "g_socket_send_messages")]
1013 fn send_messages<C: IsA<Cancellable>>(
1014 &self,
1015 messages: &mut [OutputMessage],
1016 flags: i32,
1017 cancellable: Option<&C>,
1018 ) -> Result<usize, glib::Error> {
1019 let cancellable = cancellable.map(|c| c.as_ref());
1020 unsafe {
1021 let mut error = ptr::null_mut();
1022 let count = ffi::g_socket_send_messages(
1023 self.as_ref().to_glib_none().0,
1024 messages.as_mut_ptr() as *mut _,
1025 messages.len().try_into().unwrap(),
1026 flags,
1027 cancellable.to_glib_none().0,
1028 &mut error,
1029 );
1030 if error.is_null() {
1031 Ok(count as usize)
1032 } else {
1033 Err(from_glib_full(error))
1034 }
1035 }
1036 }
1037 /// Tries to send @size bytes from @buffer to @address. If @address is
1038 /// [`None`] then the message is sent to the default receiver (set by
1039 /// g_socket_connect()).
1040 ///
1041 /// See g_socket_send() for additional information.
1042 /// ## `address`
1043 /// a #GSocketAddress, or [`None`]
1044 /// ## `buffer`
1045 /// the buffer
1046 /// containing the data to send.
1047 /// ## `cancellable`
1048 /// a `GCancellable` or [`None`]
1049 ///
1050 /// # Returns
1051 ///
1052 /// Number of bytes written (which may be less than @size), or -1
1053 /// on error
1054 #[doc(alias = "g_socket_send_to")]
1055 fn send_to<B: AsRef<[u8]>, P: IsA<SocketAddress>, C: IsA<Cancellable>>(
1056 &self,
1057 address: Option<&P>,
1058 buffer: B,
1059 cancellable: Option<&C>,
1060 ) -> Result<usize, glib::Error> {
1061 let cancellable = cancellable.map(|c| c.as_ref());
1062 let gcancellable = cancellable.to_glib_none();
1063 let (count, buffer_ptr) = {
1064 let slice = buffer.as_ref();
1065 (slice.len(), slice.as_ptr())
1066 };
1067 unsafe {
1068 let mut error = ptr::null_mut();
1069
1070 let ret = ffi::g_socket_send_to(
1071 self.as_ref().to_glib_none().0,
1072 address.map(|p| p.as_ref()).to_glib_none().0,
1073 mut_override(buffer_ptr),
1074 count,
1075 gcancellable.0,
1076 &mut error,
1077 );
1078 if error.is_null() {
1079 Ok(ret as usize)
1080 } else {
1081 Err(from_glib_full(error))
1082 }
1083 }
1084 }
1085 /// This behaves exactly the same as g_socket_send(), except that
1086 /// the choice of blocking or non-blocking behavior is determined by
1087 /// the @blocking argument rather than by @self's properties.
1088 /// ## `buffer`
1089 /// the buffer
1090 /// containing the data to send.
1091 /// ## `blocking`
1092 /// whether to do blocking or non-blocking I/O
1093 /// ## `cancellable`
1094 /// a `GCancellable` or [`None`]
1095 ///
1096 /// # Returns
1097 ///
1098 /// Number of bytes written (which may be less than @size), or -1
1099 /// on error
1100 #[doc(alias = "g_socket_send_with_blocking")]
1101 fn send_with_blocking<B: AsRef<[u8]>, C: IsA<Cancellable>>(
1102 &self,
1103 buffer: B,
1104 blocking: bool,
1105 cancellable: Option<&C>,
1106 ) -> Result<usize, glib::Error> {
1107 let cancellable = cancellable.map(|c| c.as_ref());
1108 let gcancellable = cancellable.to_glib_none();
1109 let (count, buffer_ptr) = {
1110 let slice = buffer.as_ref();
1111 (slice.len(), slice.as_ptr())
1112 };
1113 unsafe {
1114 let mut error = ptr::null_mut();
1115 let ret = ffi::g_socket_send_with_blocking(
1116 self.as_ref().to_glib_none().0,
1117 mut_override(buffer_ptr),
1118 count,
1119 blocking.into_glib(),
1120 gcancellable.0,
1121 &mut error,
1122 );
1123 if error.is_null() {
1124 Ok(ret as usize)
1125 } else {
1126 Err(from_glib_full(error))
1127 }
1128 }
1129 }
1130
1131 /// Gets the underlying OS socket descriptor.
1132 ///
1133 /// On Unix this is a socket file descriptor, and on Windows this is
1134 /// a Winsock2 `SOCKET` handle.
1135 ///
1136 /// This may be useful for doing platform specific or otherwise unusual
1137 /// operations on the socket.
1138 ///
1139 /// # Returns
1140 ///
1141 /// the file descriptor of the socket, or `-1` if the socket has not yet
1142 /// been initialised or has been closed
1143 #[cfg(unix)]
1144 #[cfg_attr(docsrs, doc(cfg(unix)))]
1145 #[doc(alias = "get_fd")]
1146 #[doc(alias = "g_socket_get_fd")]
1147 fn fd(&self) -> BorrowedFd<'_> {
1148 self.as_ref().as_fd()
1149 }
1150
1151 #[cfg(windows)]
1152 #[cfg_attr(docsrs, doc(cfg(windows)))]
1153 #[doc(alias = "get_socket")]
1154 #[doc(alias = "g_socket_get_fd")]
1155 fn socket(&self) -> BorrowedSocket<'_> {
1156 self.as_ref().as_socket()
1157 }
1158
1159 #[doc(alias = "g_socket_create_source")]
1160 fn create_source<F, C>(
1161 &self,
1162 condition: glib::IOCondition,
1163 cancellable: Option<&C>,
1164 name: Option<&str>,
1165 priority: glib::Priority,
1166 func: F,
1167 ) -> glib::Source
1168 where
1169 F: FnMut(&Self, glib::IOCondition) -> glib::ControlFlow + 'static,
1170 C: IsA<Cancellable>,
1171 {
1172 unsafe extern "C" fn trampoline<
1173 O: IsA<Socket>,
1174 F: FnMut(&O, glib::IOCondition) -> glib::ControlFlow + 'static,
1175 >(
1176 socket: *mut ffi::GSocket,
1177 condition: glib::ffi::GIOCondition,
1178 func: glib::ffi::gpointer,
1179 ) -> glib::ffi::gboolean {
1180 unsafe {
1181 let func: &RefCell<F> = &*(func as *const RefCell<F>);
1182 let mut func = func.borrow_mut();
1183 (*func)(
1184 Socket::from_glib_borrow(socket).unsafe_cast_ref(),
1185 from_glib(condition),
1186 )
1187 .into_glib()
1188 }
1189 }
1190 unsafe extern "C" fn destroy_closure<F>(ptr: glib::ffi::gpointer) {
1191 unsafe {
1192 let _ = Box::<RefCell<F>>::from_raw(ptr as *mut _);
1193 }
1194 }
1195 let cancellable = cancellable.map(|c| c.as_ref());
1196 let gcancellable = cancellable.to_glib_none();
1197 unsafe {
1198 let source = ffi::g_socket_create_source(
1199 self.as_ref().to_glib_none().0,
1200 condition.into_glib(),
1201 gcancellable.0,
1202 );
1203 let trampoline = trampoline::<Self, F> as glib::ffi::gpointer;
1204 glib::ffi::g_source_set_callback(
1205 source,
1206 Some(transmute::<
1207 glib::ffi::gpointer,
1208 unsafe extern "C" fn(glib::ffi::gpointer) -> glib::ffi::gboolean,
1209 >(trampoline)),
1210 Box::into_raw(Box::new(RefCell::new(func))) as glib::ffi::gpointer,
1211 Some(destroy_closure::<F>),
1212 );
1213 glib::ffi::g_source_set_priority(source, priority.into_glib());
1214
1215 if let Some(name) = name {
1216 glib::ffi::g_source_set_name(source, name.to_glib_none().0);
1217 }
1218
1219 from_glib_full(source)
1220 }
1221 }
1222
1223 fn create_source_future<C: IsA<Cancellable>>(
1224 &self,
1225 condition: glib::IOCondition,
1226 cancellable: Option<&C>,
1227 priority: glib::Priority,
1228 ) -> Pin<Box<dyn std::future::Future<Output = glib::IOCondition> + 'static>> {
1229 let cancellable: Option<Cancellable> = cancellable.map(|c| c.as_ref()).cloned();
1230
1231 let obj = self.clone();
1232 Box::pin(glib::SourceFuture::new(move |send| {
1233 let mut send = Some(send);
1234 obj.create_source(
1235 condition,
1236 cancellable.as_ref(),
1237 None,
1238 priority,
1239 move |_, condition| {
1240 let _ = send.take().unwrap().send(condition);
1241 glib::ControlFlow::Break
1242 },
1243 )
1244 }))
1245 }
1246
1247 fn create_source_stream<C: IsA<Cancellable>>(
1248 &self,
1249 condition: glib::IOCondition,
1250 cancellable: Option<&C>,
1251 priority: glib::Priority,
1252 ) -> Pin<Box<dyn Stream<Item = glib::IOCondition> + 'static>> {
1253 let cancellable: Option<Cancellable> = cancellable.map(|c| c.as_ref()).cloned();
1254
1255 let obj = self.clone();
1256 Box::pin(glib::SourceStream::new(move |send| {
1257 let send = Some(send);
1258 obj.create_source(
1259 condition,
1260 cancellable.as_ref(),
1261 None,
1262 priority,
1263 move |_, condition| {
1264 if send.as_ref().unwrap().unbounded_send(condition).is_err() {
1265 glib::ControlFlow::Break
1266 } else {
1267 glib::ControlFlow::Continue
1268 }
1269 },
1270 )
1271 }))
1272 }
1273}
1274
1275impl<O: IsA<Socket>> SocketExtManual for O {}
1276
1277#[cfg(all(docsrs, not(unix)))]
1278pub trait AsRawFd {
1279 fn as_raw_fd(&self) -> RawFd;
1280}
1281
1282#[cfg(all(docsrs, not(unix)))]
1283pub type RawFd = libc::c_int;
1284
1285#[cfg(all(docsrs, not(windows)))]
1286pub trait AsRawSocket {
1287 fn as_raw_socket(&self) -> RawSocket;
1288}
1289
1290#[cfg(all(docsrs, not(windows)))]
1291pub type RawSocket = *mut std::os::raw::c_void;
1292
1293#[cfg(test)]
1294mod tests {
1295 #[test]
1296 #[cfg(unix)]
1297 fn dgram_socket_messages() {
1298 use super::Socket;
1299 use crate::{Cancellable, prelude::*};
1300
1301 let addr = crate::InetSocketAddress::from_string("127.0.0.1", 28351).unwrap();
1302
1303 let out_sock = Socket::new(
1304 crate::SocketFamily::Ipv4,
1305 crate::SocketType::Datagram,
1306 crate::SocketProtocol::Udp,
1307 )
1308 .unwrap();
1309 let in_sock = Socket::new(
1310 crate::SocketFamily::Ipv4,
1311 crate::SocketType::Datagram,
1312 crate::SocketProtocol::Udp,
1313 )
1314 .unwrap();
1315 in_sock.bind(&addr, true).unwrap();
1316
1317 const DATA: [u8; std::mem::size_of::<u64>()] = 1234u64.to_be_bytes();
1318 let out_vec = DATA;
1319 let out_vecs = [super::OutputVector::new(out_vec.as_slice())];
1320 let mut out_msg = [super::OutputMessage::new(
1321 Some(&addr),
1322 out_vecs.as_slice(),
1323 &[],
1324 )];
1325 let written = super::SocketExtManual::send_messages(
1326 &out_sock,
1327 out_msg.as_mut_slice(),
1328 0,
1329 Cancellable::NONE,
1330 )
1331 .unwrap();
1332 assert_eq!(written, 1);
1333 assert_eq!(out_msg[0].bytes_sent() as usize, out_vec.len());
1334
1335 let mut in_addr = None;
1336 let mut in_vec = [0u8; DATA.len()];
1337 let mut in_vecs = [super::InputVector::new(in_vec.as_mut_slice())];
1338 let mut in_msg = [super::InputMessage::new(
1339 Some(&mut in_addr),
1340 in_vecs.as_mut_slice(),
1341 None,
1342 )];
1343 let received = super::SocketExtManual::receive_messages(
1344 &in_sock,
1345 in_msg.as_mut_slice(),
1346 0,
1347 Cancellable::NONE,
1348 )
1349 .unwrap();
1350
1351 assert_eq!(received, 1);
1352 assert_eq!(in_msg[0].bytes_received(), in_vec.len());
1353 assert_eq!(in_vec, out_vec);
1354 let in_addr = in_addr
1355 .unwrap()
1356 .downcast::<crate::InetSocketAddress>()
1357 .unwrap();
1358 assert_eq!(in_addr.address().to_str(), addr.address().to_str());
1359 }
1360}