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.
1718use bb8::PooledConnection;
19use bytes::BytesMut;
20use openssh_sftp_client::file::File;
2122use super::core::Manager;
23use super::error::parse_sftp_error;
24use crate::raw::*;
25use crate::*;
2627pub struct SftpReader {
28/// Keep the connection alive while data stream is alive.
29_conn: PooledConnection<'static, Manager>,
3031 file: File,
32 chunk: usize,
33 size: Option<usize>,
34 read: usize,
35 buf: BytesMut,
36}
3738impl SftpReader {
39pub fn new(conn: PooledConnection<'static, Manager>, file: File, size: Option<u64>) -> Self {
40Self {
41 _conn: conn,
42 file,
43 size: size.map(|v| v as usize),
44 chunk: 2 * 1024 * 1024,
45 read: 0,
46 buf: BytesMut::new(),
47 }
48 }
49}
5051impl oio::Read for SftpReader {
52async fn read(&mut self) -> Result<Buffer> {
53if self.read >= self.size.unwrap_or(usize::MAX) {
54return Ok(Buffer::new());
55 }
5657let size = if let Some(size) = self.size {
58 (size - self.read).min(self.chunk)
59 } else {
60self.chunk
61 };
62self.buf.reserve(size);
6364let Some(bytes) = self
65.file
66 .read(size as u32, self.buf.split_off(0))
67 .await
68.map_err(parse_sftp_error)?
69else {
70return Ok(Buffer::new());
71 };
7273self.read += bytes.len();
74self.buf = bytes;
75let bs = self.buf.split();
76Ok(Buffer::from(bs.freeze()))
77 }
78}