vortex_serde/io/
monoio.rs

1#![cfg(feature = "monoio")]
2
3use std::future::Future;
4use std::io;
5
6use bytes::BytesMut;
7use futures_util::FutureExt;
8use monoio::buf::IoBufMut;
9use monoio::io::{AsyncReadRent, AsyncReadRentExt, AsyncWriteRent, AsyncWriteRentExt};
10use vortex_buffer::io_buf::IoBuf;
11
12use crate::io::{VortexRead, VortexWrite};
13
14pub struct MonoAdapter<IO>(IO);
15
16impl<R: AsyncReadRent> VortexRead for MonoAdapter<R> {
17    fn read_into(&mut self, buffer: BytesMut) -> impl Future<Output = io::Result<BytesMut>> {
18        let len = buffer.len();
19        self.0
20            .read_exact(buffer.slice_mut(0..len))
21            .map(|(result, buffer)| match result {
22                Ok(_len) => Ok(buffer.into_inner()),
23                Err(e) => Err(e),
24            })
25    }
26}
27
28impl<W: AsyncWriteRent> VortexWrite for MonoAdapter<W> {
29    fn write_all<B: IoBuf>(&mut self, buffer: B) -> impl Future<Output = io::Result<B>> {
30        self.0
31            .write_all(MonoAdapter(buffer))
32            .map(|(result, buffer)| match result {
33                Ok(_len) => Ok(buffer.0),
34                Err(e) => Err(e),
35            })
36    }
37
38    fn flush(&mut self) -> impl Future<Output = io::Result<()>> {
39        self.0.flush()
40    }
41
42    fn shutdown(&mut self) -> impl Future<Output = io::Result<()>> {
43        self.0.shutdown()
44    }
45}
46
47unsafe impl<B: IoBuf> monoio::buf::IoBuf for MonoAdapter<B> {
48    fn read_ptr(&self) -> *const u8 {
49        IoBuf::read_ptr(&self.0)
50    }
51
52    fn bytes_init(&self) -> usize {
53        IoBuf::bytes_init(&self.0)
54    }
55}
vortex_serde/io/monoio.rs

vortex_serde/io/
monoio.rs
