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
14pub trait PollableInputStreamExtManual: IsA<PollableInputStream> + Sized {
15    #[doc(alias = "g_pollable_input_stream_create_source")]
16    fn create_source<F, C>(
17        &self,
18        cancellable: Option<&C>,
19        name: Option<&str>,
20        priority: glib::Priority,
21        func: F,
22    ) -> glib::Source
23    where
24        F: FnMut(&Self) -> glib::ControlFlow + 'static,
25        C: IsA<Cancellable>,
26    {
27        unsafe extern "C" fn trampoline<
28            O: IsA<PollableInputStream>,
29            F: FnMut(&O) -> glib::ControlFlow + 'static,
30        >(
31            stream: *mut ffi::GPollableInputStream,
32            func: glib::ffi::gpointer,
33        ) -> glib::ffi::gboolean {
34            let func: &RefCell<F> = &*(func as *const RefCell<F>);
35            let mut func = func.borrow_mut();
36            (*func)(PollableInputStream::from_glib_borrow(stream).unsafe_cast_ref()).into_glib()
37        }
38        unsafe extern "C" fn destroy_closure<F>(ptr: glib::ffi::gpointer) {
39            let _ = Box::<RefCell<F>>::from_raw(ptr as *mut _);
40        }
41        let cancellable = cancellable.map(|c| c.as_ref());
42        let gcancellable = cancellable.to_glib_none();
43        unsafe {
44            let source = ffi::g_pollable_input_stream_create_source(
45                self.as_ref().to_glib_none().0,
46                gcancellable.0,
47            );
48
49            let trampoline = trampoline::<Self, F> as glib::ffi::gpointer;
50            glib::ffi::g_source_set_callback(
51                source,
52                Some(transmute::<
53                    glib::ffi::gpointer,
54                    unsafe extern "C" fn(glib::ffi::gpointer) -> glib::ffi::gboolean,
55                >(trampoline)),
56                Box::into_raw(Box::new(RefCell::new(func))) as glib::ffi::gpointer,
57                Some(destroy_closure::<F>),
58            );
59            glib::ffi::g_source_set_priority(source, priority.into_glib());
60
61            if let Some(name) = name {
62                glib::ffi::g_source_set_name(source, name.to_glib_none().0);
63            }
64
65            from_glib_full(source)
66        }
67    }
68
69    fn create_source_future<C: IsA<Cancellable>>(
70        &self,
71        cancellable: Option<&C>,
72        priority: glib::Priority,
73    ) -> Pin<Box<dyn std::future::Future<Output = ()> + 'static>> {
74        let cancellable: Option<Cancellable> = cancellable.map(|c| c.as_ref()).cloned();
75
76        let obj = self.clone();
77        Box::pin(glib::SourceFuture::new(move |send| {
78            let mut send = Some(send);
79            obj.create_source(cancellable.as_ref(), None, priority, move |_| {
80                let _ = send.take().unwrap().send(());
81                glib::ControlFlow::Break
82            })
83        }))
84    }
85
86    fn create_source_stream<C: IsA<Cancellable>>(
87        &self,
88        cancellable: Option<&C>,
89        priority: glib::Priority,
90    ) -> Pin<Box<dyn Stream<Item = ()> + 'static>> {
91        let cancellable: Option<Cancellable> = cancellable.map(|c| c.as_ref()).cloned();
92
93        let obj = self.clone();
94        Box::pin(glib::SourceStream::new(move |send| {
95            obj.create_source(cancellable.as_ref(), None, priority, move |_| {
96                if send.unbounded_send(()).is_err() {
97                    glib::ControlFlow::Break
98                } else {
99                    glib::ControlFlow::Continue
100                }
101            })
102        }))
103    }
104
105    #[doc(alias = "g_pollable_input_stream_read_nonblocking")]
106    fn read_nonblocking<C: IsA<Cancellable>>(
107        &self,
108        buffer: &mut [u8],
109        cancellable: Option<&C>,
110    ) -> Result<isize, glib::Error> {
111        let cancellable = cancellable.map(|c| c.as_ref());
112        let gcancellable = cancellable.to_glib_none();
113        let count = buffer.len();
114        unsafe {
115            let mut error = ptr::null_mut();
116            let ret = ffi::g_pollable_input_stream_read_nonblocking(
117                self.as_ref().to_glib_none().0,
118                buffer.to_glib_none().0,
119                count,
120                gcancellable.0,
121                &mut error,
122            );
123            if error.is_null() {
124                Ok(ret)
125            } else {
126                Err(from_glib_full(error))
127            }
128        }
129    }
130
131    fn into_async_read(self) -> Result<InputStreamAsyncRead<Self>, Self>
132    where
133        Self: IsA<PollableInputStream>,
134    {
135        if self.can_poll() {
136            Ok(InputStreamAsyncRead(self))
137        } else {
138            Err(self)
139        }
140    }
141}
142
143impl<O: IsA<PollableInputStream>> PollableInputStreamExtManual for O {}
144
145#[derive(Debug)]
146pub struct InputStreamAsyncRead<T: IsA<PollableInputStream>>(T);
147
148impl<T: IsA<PollableInputStream>> InputStreamAsyncRead<T> {
149    pub fn into_input_stream(self) -> T {
150        self.0
151    }
152
153    pub fn input_stream(&self) -> &T {
154        &self.0
155    }
156}
157
158impl<T: IsA<PollableInputStream>> AsyncRead for InputStreamAsyncRead<T> {
159    fn poll_read(
160        self: Pin<&mut Self>,
161        cx: &mut Context,
162        buf: &mut [u8],
163    ) -> Poll<io::Result<usize>> {
164        let stream = Pin::get_ref(self.as_ref());
165        let gio_result = stream
166            .0
167            .as_ref()
168            .read_nonblocking(buf, crate::Cancellable::NONE);
169
170        match gio_result {
171            Ok(size) => Poll::Ready(Ok(size as usize)),
172            Err(err) => {
173                let kind = err
174                    .kind::<crate::IOErrorEnum>()
175                    .unwrap_or(crate::IOErrorEnum::Failed);
176                if kind == crate::IOErrorEnum::WouldBlock {
177                    let mut waker = Some(cx.waker().clone());
178                    let source = stream.0.as_ref().create_source(
179                        crate::Cancellable::NONE,
180                        None,
181                        glib::Priority::default(),
182                        move |_| {
183                            if let Some(waker) = waker.take() {
184                                waker.wake();
185                            }
186                            glib::ControlFlow::Break
187                        },
188                    );
189                    let main_context = glib::MainContext::ref_thread_default();
190                    source.attach(Some(&main_context));
191
192                    Poll::Pending
193                } else {
194                    Poll::Ready(Err(io::Error::new(io::ErrorKind::from(kind), err)))
195                }
196            }
197        }
198    }
199}
gio/pollable_input_stream.rs

gio/
pollable_input_stream.rs
