gio/
pollable_input_stream.rs
1use 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 #[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 #[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}