/* This file is part of DarkFi (https://dark.fi)
*
* Copyright (C) 2020-2026 Dyne.org foundation
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see .
*/
use std::{
collections::HashSet,
fs::{File, OpenOptions},
io::{Read, Seek, Write},
path::PathBuf,
pin::Pin,
};
use futures::{
task::{Context, Poll},
AsyncRead, AsyncSeek, AsyncWrite,
};
use smol::io::{self, SeekFrom};
/// `FileSequence` is an object that implements `AsyncRead`, `AsyncSeek`, and
/// `AsyncWrite` for an ordered list of (file path, file size).
///
/// You can use it to read and write from/to a list a file, without having to
/// manage individual file operations explicitly.
///
/// This allows seamless handling of multiple files as if they were a single
/// continuous file. It automatically opens the next file in the list when the
/// current file is exhausted.
///
/// It's also made so that files in `files` that do not exist on the filesystem
/// will get skipped, without returning an Error. All files you want to read,
/// write, and seek to should be created before using the FileSequence.
#[derive(Debug)]
pub struct FileSequence {
/// List of (file path, file size). File sizes are not the sizes of the
/// files as they currently are on the file system, but the sizes we want
files: Vec<(PathBuf, u64)>,
/// Currently opened file
current_file: Option,
/// Index of the currently opened file in the `files` vector
current_file_index: Option,
position: u64,
/// Set to `true` to automatically set the length of the file on the
/// filesystem to it's size as defined in the `files` vector, after a write
auto_set_len: bool,
}
impl FileSequence {
pub fn new(files: &[(PathBuf, u64)], auto_set_len: bool) -> Self {
Self {
files: files.to_vec(),
current_file: None,
current_file_index: None,
position: 0,
auto_set_len,
}
}
/// Update a single file size.
pub fn set_file_size(&mut self, file_index: usize, file_size: u64) {
self.files[file_index].1 = file_size;
}
/// Return `current_file`.
pub fn get_current_file(&self) -> &Option {
&self.current_file
}
/// Return `files`.
pub fn get_files(&self) -> &Vec<(PathBuf, u64)> {
&self.files
}
/// Return the combined file size of all files.
pub fn len(&self) -> u64 {
self.files.iter().map(|(_, size)| size).sum()
}
/// Return `true` if the `FileSequence` contains no file.
pub fn is_empty(&self) -> bool {
self.files.is_empty()
}
/// Return the combined file size of all files.
pub fn subset_len(&self, files: HashSet) -> u64 {
self.files.iter().filter(|(path, _)| files.contains(path)).map(|(_, size)| size).sum()
}
/// Compute the starting position of the file (in bytes) by suming up
/// the size of the previous files.
pub fn get_file_position(&self, file_index: usize) -> u64 {
let mut pos = 0;
for i in 0..file_index {
pos += self.files[i].1;
}
pos
}
/// Open the file at (`current_file_index` + 1).
/// If no file is currently open (`current_file_index` is None), it opens
/// the first file.
fn open_next_file(&mut self) -> io::Result<()> {
self.current_file = None;
self.current_file_index = match self.current_file_index {
Some(i) => Some(i + 1),
None => Some(0),
};
if self.current_file_index.unwrap() >= self.files.len() {
return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "No more files to open"))
}
let file = OpenOptions::new()
.read(true)
.write(true)
.create(false)
.open(self.files[self.current_file_index.unwrap()].0.clone())?;
self.current_file = Some(file);
Ok(())
}
/// Open the file at `file_index`.
fn open_file(&mut self, file_index: usize) -> io::Result<()> {
self.current_file = None;
self.current_file_index = Some(file_index);
let file = OpenOptions::new()
.read(true)
.write(true)
.create(false)
.open(self.files[file_index].0.clone())?;
self.current_file = Some(file);
Ok(())
}
}
impl AsyncRead for FileSequence {
fn poll_read(
self: Pin<&mut Self>,
_: &mut Context<'_>,
buf: &mut [u8],
) -> Poll> {
let this = self.get_mut();
let mut total_read = 0;
while total_read < buf.len() {
if this.current_file.is_none() {
if let Some(file_index) = this.current_file_index {
// Stop if there are no more files to read
if file_index >= this.files.len() - 1 {
return Poll::Ready(Ok(total_read));
}
let start_pos = this.get_file_position(file_index);
let file_size = this.files[file_index].1 as usize;
let file_pos = this.position - start_pos;
let space_left = file_size - file_pos as usize;
let skip_bytes = (buf.len() - total_read).min(space_left);
this.position += skip_bytes as u64;
total_read += skip_bytes;
}
// Open the next file
match this.open_next_file() {
Ok(_) => {}
Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => {
return Poll::Ready(Ok(total_read));
}
Err(e) if e.kind() == io::ErrorKind::NotFound => {
this.current_file = None;
continue; // Skip to next file
}
Err(e) => return Poll::Ready(Err(e)),
}
}
// Read from the current file
let file = this.current_file.as_mut().unwrap();
match file.read(&mut buf[total_read..]) {
Ok(bytes_read) => {
if bytes_read == 0 {
this.current_file = None; // Move to the next file
} else {
total_read += bytes_read;
this.position += bytes_read as u64;
}
}
Err(e) => return Poll::Ready(Err(e)),
}
}
Poll::Ready(Ok(total_read))
}
}
impl AsyncSeek for FileSequence {
fn poll_seek(
self: Pin<&mut Self>,
_: &mut Context<'_>,
pos: SeekFrom,
) -> Poll> {
let this = self.get_mut();
let abs_pos = match pos {
SeekFrom::Start(offset) => offset,
_ => todo!(), // TODO
};
// Determine which file to seek in
let mut file_index = 0;
let mut bytes_offset = 0;
while file_index < this.files.len() {
if bytes_offset + this.files[file_index].1 >= abs_pos {
break;
}
bytes_offset += this.files[file_index].1;
file_index += 1;
}
if file_index >= this.files.len() {
return Poll::Ready(Err(io::Error::new(
io::ErrorKind::InvalidInput,
"Seek position out of bounds",
)))
}
this.position = abs_pos; // Update FileSequence position
// Open the file
if this.current_file.is_none() ||
this.current_file_index.is_some() && this.current_file_index.unwrap() != file_index
{
match this.open_file(file_index) {
Ok(_) => {}
Err(e) if e.kind() == io::ErrorKind::NotFound => {
// If the file does not exist, return without actually seeking it
return Poll::Ready(Ok(this.position));
}
Err(e) => return Poll::Ready(Err(e)),
};
}
let file = this.current_file.as_mut().unwrap();
let file_pos = abs_pos - bytes_offset;
// Seek in the current file
match file.seek(SeekFrom::Start(file_pos)) {
Ok(_) => Poll::Ready(Ok(this.position)),
Err(e) => Poll::Ready(Err(e)),
}
}
}
impl AsyncWrite for FileSequence {
fn poll_write(
self: Pin<&mut Self>,
_: &mut Context<'_>,
buf: &[u8],
) -> Poll> {
let this = self.get_mut();
let mut total_bytes_written = 0;
let mut remaining_buf = buf;
let auto_set_len = this.auto_set_len;
let finalize_current_file = |file: &mut File, max_size: u64| {
if auto_set_len {
file.set_len(max_size)?;
}
file.flush()?;
Ok(())
};
loop {
// Ensure the current file is open
if this.current_file.is_none() {
if let Some(file_index) = this.current_file_index {
if file_index >= this.files.len() - 1 {
break; // No more files
}
if remaining_buf.is_empty() {
break; // No more data to write
}
let start_pos = this.get_file_position(file_index);
let file_size = this.files[file_index].1 as usize;
let file_pos = this.position - start_pos;
let space_left = file_size - file_pos as usize;
let skip_bytes = remaining_buf.len().min(space_left);
this.position += skip_bytes as u64;
remaining_buf = &remaining_buf[skip_bytes..]; // Update the remaining buffer
}
// Switch to the next file
match this.open_next_file() {
Ok(_) => {}
Err(e) if e.kind() == io::ErrorKind::NotFound => {
this.current_file = None;
continue; // Skip to next file
}
Err(e) => return Poll::Ready(Err(e)),
}
}
let file = this.current_file.as_mut().unwrap();
let max_size = this.files[this.current_file_index.unwrap()].1;
// Check how much space is left in the current file
let current_position = file.stream_position()?;
let space_left = max_size - current_position;
let bytes_to_write = remaining_buf.len().min(space_left as usize);
if bytes_to_write == 0 {
// Continue to the next iteration to check the new file
if let Err(e) = finalize_current_file(file, max_size) {
return Poll::Ready(Err(e));
}
this.current_file = None;
continue;
}
// Write to the current file
match file.write(&remaining_buf[..bytes_to_write]) {
Ok(bytes_written) => {
total_bytes_written += bytes_written;
this.position += bytes_written as u64;
remaining_buf = &remaining_buf[bytes_written..]; // Update the remaining buffer
if remaining_buf.is_empty() {
if let Err(e) = finalize_current_file(file, max_size) {
return Poll::Ready(Err(e));
}
break; // No more data to write
}
// We wrote to the end of this file, use new file on next iteration
if bytes_written == bytes_to_write {
if let Err(e) = finalize_current_file(file, max_size) {
return Poll::Ready(Err(e));
}
this.current_file = None;
}
}
Err(e) => return Poll::Ready(Err(e)), // Return error if write fails
}
}
Poll::Ready(Ok(total_bytes_written))
}
fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> {
Poll::Ready(Ok(())) // TODO
}
fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> {
let this = self.get_mut();
if let Some(file) = this.current_file.take() {
match file.sync_all() {
Ok(()) => Poll::Ready(Ok(())),
Err(e) => Poll::Ready(Err(e)),
}
} else {
Poll::Ready(Ok(())) // No file to close
}
}
}
impl Clone for FileSequence {
fn clone(&self) -> Self {
Self {
files: self.files.clone(),
current_file: None,
current_file_index: None,
position: 0,
auto_set_len: self.auto_set_len,
}
}
}