opendal/services/hdfs/
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 bytes::BytesMut;
19use futures::AsyncReadExt;
20use hdrs::AsyncFile;
21use tokio::io::ReadBuf;
22
23use crate::raw::*;
24use crate::*;
25
26pub struct HdfsReader<F> {
27    f: F,
28    read: usize,
29    size: usize,
30    buf_size: usize,
31    buf: BytesMut,
32}
33
34impl<F> HdfsReader<F> {
35    pub fn new(f: F, size: usize) -> Self {
36        Self {
37            f,
38            read: 0,
39            size,
40            // Use 2 MiB as default value.
41            buf_size: 2 * 1024 * 1024,
42            buf: BytesMut::new(),
43        }
44    }
45}
46
47impl oio::Read for HdfsReader<AsyncFile> {
48    async fn read(&mut self) -> Result<Buffer> {
49        if self.read >= self.size {
50            return Ok(Buffer::new());
51        }
52
53        let size = (self.size - self.read).min(self.buf_size);
54        self.buf.reserve(size);
55
56        let buf = &mut self.buf.spare_capacity_mut()[..size];
57        let mut read_buf: ReadBuf = ReadBuf::uninit(buf);
58
59        // SAFETY: Read at most `limit` bytes into `read_buf`.
60        unsafe {
61            read_buf.assume_init(size);
62        }
63
64        let n = self
65            .f
66            .read(read_buf.initialize_unfilled())
67            .await
68            .map_err(new_std_io_error)?;
69        read_buf.advance(n);
70        self.read += n;
71
72        // Safety: We make sure that bs contains `n` more bytes.
73        let filled = read_buf.filled().len();
74        unsafe { self.buf.set_len(filled) }
75
76        let frozen = self.buf.split().freeze();
77        Ok(Buffer::from(frozen))
78    }
79}
opendal/services/hdfs/reader.rs

opendal/services/hdfs/
reader.rs
