gio: support writev in stream impls for io::Write, AsyncWrite

Author: jf2048

gio/src/output_stream.rs

@@ -668,6 +668,30 @@ impl<T: IsA<OutputStream>> io::Write for OutputStreamWrite<T> {
         to_std_io_result(result)
     }
 
+    fn write_all(&mut self, buf: &[u8]) -> io::Result<()> {
+        let result = self
+            .0
+            .as_ref()
+            .write_all(buf, crate::Cancellable::NONE)
+            .and_then(|(_, e)| e.map(Err).unwrap_or(Ok(())));
+        to_std_io_result(result)
+    }
+
+    #[cfg(any(feature = "v2_60", feature = "dox"))]
+    #[cfg_attr(feature = "dox", doc(cfg(feature = "v2_60")))]
+    fn write_vectored(&mut self, bufs: &[io::IoSlice<'_>]) -> io::Result<usize> {
+        let vectors = bufs
+            .iter()
+            .map(|v| OutputVector::new(&**v))
+            .collect::<smallvec::SmallVec<[_; 2]>>();
+        let result = self
+            .0
+            .as_ref()
+            .writev(&vectors, crate::Cancellable::NONE)
+            .map(|size| size as usize);
+        to_std_io_result(result)
+    }
+
     fn flush(&mut self) -> io::Result<()> {
         let gio_result = self.0.as_ref().flush(crate::Cancellable::NONE);
         to_std_io_result(gio_result)

gio/src/pollable_output_stream.rs

@@ -236,6 +236,51 @@ impl<T: IsA<PollableOutputStream>> AsyncWrite for OutputStreamAsyncWrite<T> {
         }
     }
 
+    #[cfg(any(feature = "v2_60", feature = "dox"))]
+    #[cfg_attr(feature = "dox", doc(cfg(feature = "v2_60")))]
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[io::IoSlice<'_>],
+    ) -> Poll<io::Result<usize>> {
+        let stream = Pin::get_ref(self.as_ref());
+        let vectors = bufs
+            .iter()
+            .map(|v| OutputVector::new(&**v))
+            .collect::<smallvec::SmallVec<[_; 2]>>();
+        let gio_result = stream
+            .0
+            .as_ref()
+            .writev_nonblocking(&vectors, crate::Cancellable::NONE);
+
+        match gio_result {
+            Ok((PollableReturn::Ok, size)) => Poll::Ready(Ok(size as usize)),
+            Ok((PollableReturn::WouldBlock, _)) => {
+                let mut waker = Some(cx.waker().clone());
+                let source = stream.0.as_ref().create_source(
+                    crate::Cancellable::NONE,
+                    None,
+                    glib::PRIORITY_DEFAULT,
+                    move |_| {
+                        if let Some(waker) = waker.take() {
+                            waker.wake();
+                        }
+                        glib::Continue(false)
+                    },
+                );
+                let main_context = glib::MainContext::ref_thread_default();
+                source.attach(Some(&main_context));
+
+                Poll::Pending
+            }
+            Ok((_, _)) => unreachable!(),
+            Err(err) => Poll::Ready(Err(io::Error::new(
+                io::ErrorKind::from(err.kind::<crate::IOErrorEnum>().unwrap()),
+                err,
+            ))),
+        }
+    }
+
     fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
         let stream = unsafe { Pin::get_unchecked_mut(self) };