move file_chunker to tools

This commit is contained in:
Dietmar Maurer 2018-12-15 11:14:41 +01:00
parent 48e8689505
commit 43eeef28d1
2 changed files with 70 additions and 75 deletions

View File

@ -2,19 +2,12 @@ extern crate apitest;
use failure::*; use failure::*;
use std::collections::HashMap; use apitest::tools;
use std::fs::File;
use std::io::Read;
use std::io::ErrorKind;
use std::io::prelude::*;
use std::iter::Iterator;
use apitest::cli::command::*; use apitest::cli::command::*;
use apitest::api::schema::*; use apitest::api::schema::*;
use apitest::api::router::*; use apitest::api::router::*;
use apitest::backup::chunk_store::*; use apitest::backup::chunk_store::*;
use serde_json::{json, Value}; use serde_json::{Value};
use std::path::{Path, PathBuf};
use apitest::config::datastore; use apitest::config::datastore;
@ -23,71 +16,6 @@ fn required_string_param<'a>(param: &'a Value, name: &str) -> &'a str {
} }
// Note: We cannot implement an Iterator, because Iterators cannot
// return a borrowed buffer ref (we want zero-copy)
fn file_chunker<C, R>(
mut file: R,
chunk_size: usize,
chunk_cb: C
) -> Result<(), Error>
where C: Fn(usize, &[u8]) -> Result<bool, Error>,
R: Read,
{
const read_buffer_size: usize = 4*1024*1024; // 4M
if chunk_size > read_buffer_size { bail!("chunk size too large!"); }
let mut buf = vec![0u8; read_buffer_size];
let mut pos = 0;
let mut file_pos = 0;
loop {
let mut eof = false;
let mut tmp = &mut buf[..];
// try to read large portions, at least chunk_size
while pos < chunk_size {
match file.read(tmp) {
Ok(0) => { eof = true; break; },
Ok(n) => {
pos += n;
if pos > chunk_size { break; }
tmp = &mut tmp[n..];
}
Err(ref e) if e.kind() == ErrorKind::Interrupted => { /* try again */ }
Err(e) => bail!("read error - {}", e.to_string()),
}
}
println!("READ {} {}", pos, eof);
let mut start = 0;
while start + chunk_size <= pos {
if !(chunk_cb)(file_pos, &buf[start..start+chunk_size])? { break; }
file_pos += chunk_size;
start += chunk_size;
}
if eof {
if start < pos {
(chunk_cb)(file_pos, &buf[start..pos])?;
//file_pos += pos - start;
}
break;
} else {
let rest = pos - start;
if rest > 0 {
let ptr = buf.as_mut_ptr();
unsafe { std::ptr::copy_nonoverlapping(ptr.add(start), ptr, rest); }
pos = rest;
} else {
pos = 0;
}
}
}
Ok(())
}
fn backup_file(param: Value, _info: &ApiMethod) -> Result<Value, Error> { fn backup_file(param: Value, _info: &ApiMethod) -> Result<Value, Error> {
let filename = required_string_param(&param, "filename"); let filename = required_string_param(&param, "filename");
@ -105,7 +33,7 @@ fn backup_file(param: Value, _info: &ApiMethod) -> Result<Value, Error> {
let file = std::fs::File::open(filename)?; let file = std::fs::File::open(filename)?;
file_chunker(file, 64*1024, |pos, chunk| { tools::file_chunker(file, 64*1024, |pos, chunk| {
println!("CHUNK {} {}", pos, chunk.len()); println!("CHUNK {} {}", pos, chunk.len());
Ok(true) Ok(true)
})?; })?;

View File

@ -5,6 +5,8 @@ use nix::sys::stat;
use std::fs::File; use std::fs::File;
use std::io::Write; use std::io::Write;
use std::path::Path; use std::path::Path;
use std::io::Read;
use std::io::ErrorKind;
pub fn file_set_contents<P: AsRef<Path>>( pub fn file_set_contents<P: AsRef<Path>>(
path: P, path: P,
@ -50,3 +52,68 @@ pub fn file_set_contents<P: AsRef<Path>>(
Ok(()) Ok(())
} }
// Note: We cannot implement an Iterator, because Iterators cannot
// return a borrowed buffer ref (we want zero-copy)
pub fn file_chunker<C, R>(
mut file: R,
chunk_size: usize,
chunk_cb: C
) -> Result<(), Error>
where C: Fn(usize, &[u8]) -> Result<bool, Error>,
R: Read,
{
const READ_BUFFER_SIZE: usize = 4*1024*1024; // 4M
if chunk_size > READ_BUFFER_SIZE { bail!("chunk size too large!"); }
let mut buf = vec![0u8; READ_BUFFER_SIZE];
let mut pos = 0;
let mut file_pos = 0;
loop {
let mut eof = false;
let mut tmp = &mut buf[..];
// try to read large portions, at least chunk_size
while pos < chunk_size {
match file.read(tmp) {
Ok(0) => { eof = true; break; },
Ok(n) => {
pos += n;
if pos > chunk_size { break; }
tmp = &mut tmp[n..];
}
Err(ref e) if e.kind() == ErrorKind::Interrupted => { /* try again */ }
Err(e) => bail!("read error - {}", e.to_string()),
}
}
println!("READ {} {}", pos, eof);
let mut start = 0;
while start + chunk_size <= pos {
if !(chunk_cb)(file_pos, &buf[start..start+chunk_size])? { break; }
file_pos += chunk_size;
start += chunk_size;
}
if eof {
if start < pos {
(chunk_cb)(file_pos, &buf[start..pos])?;
//file_pos += pos - start;
}
break;
} else {
let rest = pos - start;
if rest > 0 {
let ptr = buf.as_mut_ptr();
unsafe { std::ptr::copy_nonoverlapping(ptr.add(start), ptr, rest); }
pos = rest;
} else {
pos = 0;
}
}
}
Ok(())
}