opendal_core/blocking/read/
std_reader.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::io;
19use std::io::BufRead;
20use std::io::Read;
21use std::io::Seek;
22use std::io::SeekFrom;
23
24use futures::AsyncBufReadExt;
25use futures::AsyncReadExt;
26use futures::AsyncSeekExt;
27
28use crate::*;
29
30/// `StdReader` adapts a [`crate::blocking::Reader`] to [`Read`], [`Seek`], and
31/// [`BufRead`].
32///
33/// Users can use this adapter in cases where they need to use [`Read`] or [`BufRead`] trait.
34///
35/// StdReader also implements [`Send`] and [`Sync`].
36pub struct StdReader {
37    handle: tokio::runtime::Handle,
38    r: Option<FuturesAsyncReader>,
39}
40
41impl StdReader {
42    /// NOTE: don't allow users to create StdReader directly.
43    #[inline]
44    pub(super) fn new(handle: tokio::runtime::Handle, r: FuturesAsyncReader) -> Self {
45        Self { handle, r: Some(r) }
46    }
47
48    /// Read at most `size` bytes and return them as an OpenDAL [`Buffer`].
49    ///
50    /// This method preserves the underlying buffer storage and does not copy
51    /// the payload. It can return fewer bytes than requested, including an
52    /// empty buffer at EOF.
53    pub fn read_buffer(&mut self, size: usize) -> io::Result<Buffer> {
54        let Some(r) = self.r.as_mut() else {
55            return Err(Error::new(ErrorKind::Unexpected, "reader has been dropped").into());
56        };
57
58        self.handle.block_on(r.read_buffer(size))
59    }
60
61    /// Read all remaining bytes and return them as an OpenDAL [`Buffer`].
62    ///
63    /// This method preserves the underlying buffer storage and only allocates
64    /// metadata when multiple buffers must be combined.
65    pub fn read_to_end_buffer(&mut self) -> io::Result<Buffer> {
66        let Some(r) = self.r.as_mut() else {
67            return Err(Error::new(ErrorKind::Unexpected, "reader has been dropped").into());
68        };
69
70        self.handle.block_on(r.read_to_end_buffer())
71    }
72}
73
74impl BufRead for StdReader {
75    fn fill_buf(&mut self) -> io::Result<&[u8]> {
76        let Some(r) = self.r.as_mut() else {
77            return Err(Error::new(ErrorKind::Unexpected, "reader has been dropped").into());
78        };
79
80        self.handle.block_on(r.fill_buf())
81    }
82
83    fn consume(&mut self, amt: usize) {
84        let Some(r) = self.r.as_mut() else {
85            return;
86        };
87
88        r.consume_unpin(amt);
89    }
90}
91
92impl Read for StdReader {
93    #[inline]
94    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
95        let Some(r) = self.r.as_mut() else {
96            return Err(Error::new(ErrorKind::Unexpected, "reader has been dropped").into());
97        };
98
99        self.handle.block_on(r.read(buf))
100    }
101}
102
103impl Seek for StdReader {
104    #[inline]
105    fn seek(&mut self, pos: SeekFrom) -> io::Result<u64> {
106        let Some(r) = self.r.as_mut() else {
107            return Err(Error::new(ErrorKind::Unexpected, "reader has been dropped").into());
108        };
109
110        self.handle.block_on(r.seek(pos))
111    }
112}
113
114/// Make sure the inner reader is dropped in async context.
115impl Drop for StdReader {
116    fn drop(&mut self) {
117        if let Some(v) = self.r.take() {
118            self.handle.block_on(async move { drop(v) });
119        }
120    }
121}
opendal_core/blocking/read/std_reader.rs

opendal_core/blocking/read/
std_reader.rs
