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

gio/
pollable_input_stream.rs
