gio/
pollable_input_stream.rs

1// Take a look at the license at the top of the repository in the LICENSE file.
2
3use std::{cell::RefCell, io, mem::transmute, pin::Pin, ptr};
4
5use futures_core::{
6    stream::Stream,
7    task::{Context, Poll},
8};
9use futures_io::AsyncRead;
10use glib::{prelude::*, translate::*};
11
12use crate::{ffi, prelude::*, Cancellable, PollableInputStream};
13
14mod sealed {
15    pub trait Sealed {}
16    impl<T: super::IsA<super::PollableInputStream>> Sealed for T {}
17}
18
19pub trait PollableInputStreamExtManual: sealed::Sealed + IsA<PollableInputStream> + Sized {
20    /// Creates a #GSource that triggers when @self can be read, or
21    /// @cancellable is triggered or an error occurs. The callback on the
22    /// source is of the #GPollableSourceFunc type.
23    ///
24    /// As with g_pollable_input_stream_is_readable(), it is possible that
25    /// the stream may not actually be readable even after the source
26    /// triggers, so you should use g_pollable_input_stream_read_nonblocking()
27    /// rather than g_input_stream_read() from the callback.
28    ///
29    /// The behaviour of this method is undefined if
30    /// g_pollable_input_stream_can_poll() returns [`false`] for @self.
31    /// ## `cancellable`
32    /// a #GCancellable, or [`None`]
33    ///
34    /// # Returns
35    ///
36    /// a new #GSource
37    #[doc(alias = "g_pollable_input_stream_create_source")]
38    fn create_source<F, C>(
39        &self,
40        cancellable: Option<&C>,
41        name: Option<&str>,
42        priority: glib::Priority,
43        func: F,
44    ) -> glib::Source
45    where
46        F: FnMut(&Self) -> glib::ControlFlow + 'static,
47        C: IsA<Cancellable>,
48    {
49        unsafe extern "C" fn trampoline<
50            O: IsA<PollableInputStream>,
51            F: FnMut(&O) -> glib::ControlFlow + 'static,
52        >(
53            stream: *mut ffi::GPollableInputStream,
54            func: glib::ffi::gpointer,
55        ) -> glib::ffi::gboolean {
56            let func: &RefCell<F> = &*(func as *const RefCell<F>);
57            let mut func = func.borrow_mut();
58            (*func)(PollableInputStream::from_glib_borrow(stream).unsafe_cast_ref()).into_glib()
59        }
60        unsafe extern "C" fn destroy_closure<F>(ptr: glib::ffi::gpointer) {
61            let _ = Box::<RefCell<F>>::from_raw(ptr as *mut _);
62        }
63        let cancellable = cancellable.map(|c| c.as_ref());
64        let gcancellable = cancellable.to_glib_none();
65        unsafe {
66            let source = ffi::g_pollable_input_stream_create_source(
67                self.as_ref().to_glib_none().0,
68                gcancellable.0,
69            );
70
71            let trampoline = trampoline::<Self, F> as glib::ffi::gpointer;
72            glib::ffi::g_source_set_callback(
73                source,
74                Some(transmute::<
75                    glib::ffi::gpointer,
76                    unsafe extern "C" fn(glib::ffi::gpointer) -> glib::ffi::gboolean,
77                >(trampoline)),
78                Box::into_raw(Box::new(RefCell::new(func))) as glib::ffi::gpointer,
79                Some(destroy_closure::<F>),
80            );
81            glib::ffi::g_source_set_priority(source, priority.into_glib());
82
83            if let Some(name) = name {
84                glib::ffi::g_source_set_name(source, name.to_glib_none().0);
85            }
86
87            from_glib_full(source)
88        }
89    }
90
91    fn create_source_future<C: IsA<Cancellable>>(
92        &self,
93        cancellable: Option<&C>,
94        priority: glib::Priority,
95    ) -> Pin<Box<dyn std::future::Future<Output = ()> + 'static>> {
96        let cancellable: Option<Cancellable> = cancellable.map(|c| c.as_ref()).cloned();
97
98        let obj = self.clone();
99        Box::pin(glib::SourceFuture::new(move |send| {
100            let mut send = Some(send);
101            obj.create_source(cancellable.as_ref(), None, priority, move |_| {
102                let _ = send.take().unwrap().send(());
103                glib::ControlFlow::Break
104            })
105        }))
106    }
107
108    fn create_source_stream<C: IsA<Cancellable>>(
109        &self,
110        cancellable: Option<&C>,
111        priority: glib::Priority,
112    ) -> Pin<Box<dyn Stream<Item = ()> + 'static>> {
113        let cancellable: Option<Cancellable> = cancellable.map(|c| c.as_ref()).cloned();
114
115        let obj = self.clone();
116        Box::pin(glib::SourceStream::new(move |send| {
117            obj.create_source(cancellable.as_ref(), None, priority, move |_| {
118                if send.unbounded_send(()).is_err() {
119                    glib::ControlFlow::Break
120                } else {
121                    glib::ControlFlow::Continue
122                }
123            })
124        }))
125    }
126
127    /// Attempts to read up to @count bytes from @self into @buffer, as
128    /// with g_input_stream_read(). If @self is not currently readable,
129    /// this will immediately return [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock], and you can
130    /// use g_pollable_input_stream_create_source() to create a #GSource
131    /// that will be triggered when @self is readable.
132    ///
133    /// Note that since this method never blocks, you cannot actually
134    /// use @cancellable to cancel it. However, it will return an error
135    /// if @cancellable has already been cancelled when you call, which
136    /// may happen if you call this method after a source triggers due
137    /// to having been cancelled.
138    ///
139    /// The behaviour of this method is undefined if
140    /// g_pollable_input_stream_can_poll() returns [`false`] for @self.
141    /// ## `cancellable`
142    /// a #GCancellable, or [`None`]
143    ///
144    /// # Returns
145    ///
146    /// the number of bytes read, or -1 on error (including
147    ///   [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock]).
148    ///
149    /// ## `buffer`
150    /// a
151    ///     buffer to read data into (which should be at least @count bytes long).
152    #[doc(alias = "g_pollable_input_stream_read_nonblocking")]
153    fn read_nonblocking<C: IsA<Cancellable>>(
154        &self,
155        buffer: &mut [u8],
156        cancellable: Option<&C>,
157    ) -> Result<isize, glib::Error> {
158        let cancellable = cancellable.map(|c| c.as_ref());
159        let gcancellable = cancellable.to_glib_none();
160        let count = buffer.len();
161        unsafe {
162            let mut error = ptr::null_mut();
163            let ret = ffi::g_pollable_input_stream_read_nonblocking(
164                self.as_ref().to_glib_none().0,
165                buffer.to_glib_none().0,
166                count,
167                gcancellable.0,
168                &mut error,
169            );
170            if error.is_null() {
171                Ok(ret)
172            } else {
173                Err(from_glib_full(error))
174            }
175        }
176    }
177
178    fn into_async_read(self) -> Result<InputStreamAsyncRead<Self>, Self>
179    where
180        Self: IsA<PollableInputStream>,
181    {
182        if self.can_poll() {
183            Ok(InputStreamAsyncRead(self))
184        } else {
185            Err(self)
186        }
187    }
188}
189
190impl<O: IsA<PollableInputStream>> PollableInputStreamExtManual for O {}
191
192#[derive(Debug)]
193pub struct InputStreamAsyncRead<T: IsA<PollableInputStream>>(T);
194
195impl<T: IsA<PollableInputStream>> InputStreamAsyncRead<T> {
196    pub fn into_input_stream(self) -> T {
197        self.0
198    }
199
200    pub fn input_stream(&self) -> &T {
201        &self.0
202    }
203}
204
205impl<T: IsA<PollableInputStream>> AsyncRead for InputStreamAsyncRead<T> {
206    fn poll_read(
207        self: Pin<&mut Self>,
208        cx: &mut Context,
209        buf: &mut [u8],
210    ) -> Poll<io::Result<usize>> {
211        let stream = Pin::get_ref(self.as_ref());
212        let gio_result = stream
213            .0
214            .as_ref()
215            .read_nonblocking(buf, crate::Cancellable::NONE);
216
217        match gio_result {
218            Ok(size) => Poll::Ready(Ok(size as usize)),
219            Err(err) => {
220                let kind = err
221                    .kind::<crate::IOErrorEnum>()
222                    .unwrap_or(crate::IOErrorEnum::Failed);
223                if kind == crate::IOErrorEnum::WouldBlock {
224                    let mut waker = Some(cx.waker().clone());
225                    let source = stream.0.as_ref().create_source(
226                        crate::Cancellable::NONE,
227                        None,
228                        glib::Priority::default(),
229                        move |_| {
230                            if let Some(waker) = waker.take() {
231                                waker.wake();
232                            }
233                            glib::ControlFlow::Break
234                        },
235                    );
236                    let main_context = glib::MainContext::ref_thread_default();
237                    source.attach(Some(&main_context));
238
239                    Poll::Pending
240                } else {
241                    Poll::Ready(Err(io::Error::new(io::ErrorKind::from(kind), err)))
242                }
243            }
244        }
245    }
246}