♻️ Extract file_server module
parent
6d3a5ac9ba
commit
9f399b0440
|
@ -1,9 +1,5 @@
|
||||||
use std::{
|
use std::{
|
||||||
cmp::{min, max},
|
|
||||||
convert::{Infallible, TryInto},
|
|
||||||
error::Error,
|
error::Error,
|
||||||
io::SeekFrom,
|
|
||||||
path::PathBuf,
|
|
||||||
sync::Arc,
|
sync::Arc,
|
||||||
time::Duration,
|
time::Duration,
|
||||||
};
|
};
|
||||||
|
@ -11,253 +7,15 @@ use std::{
|
||||||
use hyper::{
|
use hyper::{
|
||||||
StatusCode,
|
StatusCode,
|
||||||
};
|
};
|
||||||
use lazy_static::*;
|
use reqwest::Client;
|
||||||
use regex::Regex;
|
|
||||||
use reqwest::{
|
|
||||||
Body,
|
|
||||||
Client,
|
|
||||||
};
|
|
||||||
use tokio::{
|
use tokio::{
|
||||||
fs::{
|
|
||||||
File,
|
|
||||||
read_dir,
|
|
||||||
ReadDir,
|
|
||||||
},
|
|
||||||
io::AsyncReadExt,
|
|
||||||
sync::mpsc::{
|
|
||||||
channel,
|
|
||||||
},
|
|
||||||
time::delay_for,
|
time::delay_for,
|
||||||
};
|
};
|
||||||
|
|
||||||
use ptth::http_serde;
|
use ptth::http_serde;
|
||||||
|
|
||||||
fn parse_range_header (range_str: &str) -> (Option <u64>, Option <u64>) {
|
|
||||||
lazy_static! {
|
|
||||||
static ref RE: Regex = Regex::new (r"^bytes=(\d*)-(\d*)$").expect ("Couldn't compile regex for Range header");
|
|
||||||
}
|
|
||||||
|
|
||||||
println! ("{}", range_str);
|
|
||||||
|
|
||||||
let caps = match RE.captures (range_str) {
|
|
||||||
Some (x) => x,
|
|
||||||
_ => return (None, None),
|
|
||||||
};
|
|
||||||
let start = caps.get (1).map (|x| x.as_str ());
|
|
||||||
let end = caps.get (2).map (|x| x.as_str ());
|
|
||||||
|
|
||||||
let start = start.map (|x| u64::from_str_radix (x, 10).ok ()).flatten ();
|
|
||||||
let end = end.map (|x| u64::from_str_radix (x, 10).ok ()).flatten ();
|
|
||||||
|
|
||||||
(start, end)
|
|
||||||
}
|
|
||||||
|
|
||||||
const SERVER_URL: &str = "http://127.0.0.1:4000";
|
const SERVER_URL: &str = "http://127.0.0.1:4000";
|
||||||
|
|
||||||
async fn serve_dir (mut dir: ReadDir) -> http_serde::Response {
|
|
||||||
let (tx, rx) = channel (2);
|
|
||||||
|
|
||||||
tokio::spawn (async move {
|
|
||||||
let mut tx = tx;
|
|
||||||
|
|
||||||
tx.send (Ok::<_, Infallible> (String::from ("<style>body {font-family:sans;}</style><ul>").into_bytes ())).await.unwrap ();
|
|
||||||
|
|
||||||
while let Ok (entry) = dir.next_entry ().await {
|
|
||||||
let entry: tokio::fs::DirEntry = match entry {
|
|
||||||
Some (x) => x,
|
|
||||||
None => break,
|
|
||||||
};
|
|
||||||
|
|
||||||
let trailing_slash = match entry.file_type ().await {
|
|
||||||
Ok (t) => if t.is_dir () {
|
|
||||||
"/"
|
|
||||||
}
|
|
||||||
else {
|
|
||||||
""
|
|
||||||
},
|
|
||||||
Err (_) => "",
|
|
||||||
};
|
|
||||||
|
|
||||||
let file_name = match entry.file_name ().into_string () {
|
|
||||||
Ok (x) => x,
|
|
||||||
Err (_) => {
|
|
||||||
let s = format! ("<li>(file_name error)</li>\n");
|
|
||||||
if tx.send (Ok::<_, Infallible> (s.into_bytes ())).await.is_err ()
|
|
||||||
{
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
};
|
|
||||||
|
|
||||||
use std::borrow::Cow;
|
|
||||||
use percent_encoding::*;
|
|
||||||
|
|
||||||
let percent_file_name: Cow <str> = utf8_percent_encode (&file_name, CONTROLS).into ();
|
|
||||||
|
|
||||||
let s = format! ("<li><a href=\"{}{}\">{}{}</a></li>\n", percent_file_name, trailing_slash, file_name, trailing_slash);
|
|
||||||
if tx.send (Ok::<_, Infallible> (s.into_bytes ())).await.is_err ()
|
|
||||||
{
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
tx.send (Ok::<_, Infallible> (String::from ("</ul>").into_bytes ())).await.unwrap ();
|
|
||||||
});
|
|
||||||
|
|
||||||
let mut response = http_serde::Response::default ();
|
|
||||||
|
|
||||||
response.header ("content-type".into (), String::from ("text/html").into_bytes ());
|
|
||||||
response.body (Body::wrap_stream (rx));
|
|
||||||
|
|
||||||
response
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn serve_file (
|
|
||||||
mut f: File,
|
|
||||||
should_send_body: bool,
|
|
||||||
range_start: Option <u64>,
|
|
||||||
range_end: Option <u64>
|
|
||||||
) -> http_serde::Response {
|
|
||||||
let (tx, rx) = channel (2);
|
|
||||||
let body = if should_send_body {
|
|
||||||
Some (Body::wrap_stream (rx))
|
|
||||||
}
|
|
||||||
else {
|
|
||||||
None
|
|
||||||
};
|
|
||||||
|
|
||||||
let file_md = f.metadata ().await.unwrap ();
|
|
||||||
let file_len = file_md.len ();
|
|
||||||
|
|
||||||
let start = range_start.unwrap_or (0);
|
|
||||||
let end = range_end.unwrap_or (file_len);
|
|
||||||
|
|
||||||
let start = max (0, min (start, file_len));
|
|
||||||
let end = max (0, min (end, file_len));
|
|
||||||
|
|
||||||
f.seek (SeekFrom::Start (start)).await.unwrap ();
|
|
||||||
|
|
||||||
println! ("Serving range {}-{}", start, end);
|
|
||||||
|
|
||||||
if should_send_body {
|
|
||||||
tokio::spawn (async move {
|
|
||||||
//println! ("Opening file {:?}", path);
|
|
||||||
|
|
||||||
let mut tx = tx;
|
|
||||||
//let mut bytes_sent = 0;
|
|
||||||
let mut bytes_left = end - start;
|
|
||||||
|
|
||||||
loop {
|
|
||||||
let mut buffer = vec! [0u8; 65_536];
|
|
||||||
let bytes_read: u64 = f.read (&mut buffer).await.unwrap ().try_into ().unwrap ();
|
|
||||||
|
|
||||||
let bytes_read = min (bytes_left, bytes_read);
|
|
||||||
|
|
||||||
buffer.truncate (bytes_read.try_into ().unwrap ());
|
|
||||||
|
|
||||||
if bytes_read == 0 {
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
|
|
||||||
if tx.send (Ok::<_, Infallible> (buffer)).await.is_err () {
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
|
|
||||||
bytes_left -= bytes_read;
|
|
||||||
if bytes_left == 0 {
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
|
|
||||||
//bytes_sent += bytes_read;
|
|
||||||
//println! ("Sent {} bytes", bytes_sent);
|
|
||||||
|
|
||||||
//delay_for (Duration::from_millis (50)).await;
|
|
||||||
}
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
let mut response = http_serde::Response::default ();
|
|
||||||
|
|
||||||
response.header (String::from ("accept-ranges"), b"bytes".to_vec ());
|
|
||||||
|
|
||||||
if range_start.is_none () && range_end.is_none () {
|
|
||||||
response.status_code (http_serde::StatusCode::Ok);
|
|
||||||
response.header (String::from ("content-length"), end.to_string ().into_bytes ());
|
|
||||||
}
|
|
||||||
else {
|
|
||||||
response.status_code (http_serde::StatusCode::PartialContent);
|
|
||||||
response.header (String::from ("content-range"), format! ("bytes {}-{}/{}", start, end - 1, end).into_bytes ());
|
|
||||||
}
|
|
||||||
|
|
||||||
if let Some (body) = body {
|
|
||||||
response.body (body);
|
|
||||||
}
|
|
||||||
|
|
||||||
response
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn serve_error (
|
|
||||||
status_code: http_serde::StatusCode,
|
|
||||||
msg: String
|
|
||||||
)
|
|
||||||
-> http_serde::Response
|
|
||||||
{
|
|
||||||
let mut resp = http_serde::Response::default ();
|
|
||||||
resp.status_code (status_code)
|
|
||||||
.body (msg.into ());
|
|
||||||
resp
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn serve_all (
|
|
||||||
parts: http_serde::RequestParts
|
|
||||||
)
|
|
||||||
-> http_serde::Response
|
|
||||||
{
|
|
||||||
println! ("Client requested {}", parts.uri);
|
|
||||||
let mut range_start = None;
|
|
||||||
let mut range_end = None;
|
|
||||||
|
|
||||||
for (k, v) in parts.headers.iter () {
|
|
||||||
let v = std::str::from_utf8 (v).unwrap ();
|
|
||||||
//println! ("{}: {}", k, v);
|
|
||||||
|
|
||||||
if k == "range" {
|
|
||||||
let (start, end) = parse_range_header (v);
|
|
||||||
range_start = start;
|
|
||||||
range_end = end;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
let should_send_body = match &parts.method {
|
|
||||||
http_serde::Method::Get => true,
|
|
||||||
_ => false,
|
|
||||||
};
|
|
||||||
|
|
||||||
use percent_encoding::*;
|
|
||||||
|
|
||||||
let encoded_path = &parts.uri [1..];
|
|
||||||
let path = percent_decode (encoded_path.as_bytes ()).decode_utf8 ().unwrap ();
|
|
||||||
|
|
||||||
let mut full_path = PathBuf::from ("/home/user");
|
|
||||||
full_path.push (&*path);
|
|
||||||
|
|
||||||
if let Ok (dir) = read_dir (&full_path).await {
|
|
||||||
serve_dir (dir).await
|
|
||||||
}
|
|
||||||
else if let Ok (file) = File::open (&full_path).await {
|
|
||||||
serve_file (
|
|
||||||
file,
|
|
||||||
should_send_body,
|
|
||||||
range_start,
|
|
||||||
range_end
|
|
||||||
).await
|
|
||||||
}
|
|
||||||
else {
|
|
||||||
serve_error (http_serde::StatusCode::NotFound, "404 Not Found".into ()).await
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn handle_req_resp (
|
async fn handle_req_resp (
|
||||||
client: &Client,
|
client: &Client,
|
||||||
req_resp: reqwest::Response
|
req_resp: reqwest::Response
|
||||||
|
@ -269,8 +27,6 @@ async fn handle_req_resp (
|
||||||
return Ok (());
|
return Ok (());
|
||||||
}
|
}
|
||||||
|
|
||||||
println! ("Step 3");
|
|
||||||
|
|
||||||
let body = req_resp.bytes ().await?;
|
let body = req_resp.bytes ().await?;
|
||||||
let wrapped_req: http_serde::WrappedRequest = match rmp_serde::from_read_ref (&body)
|
let wrapped_req: http_serde::WrappedRequest = match rmp_serde::from_read_ref (&body)
|
||||||
{
|
{
|
||||||
|
@ -280,7 +36,7 @@ async fn handle_req_resp (
|
||||||
|
|
||||||
let (req_id, parts) = (wrapped_req.id, wrapped_req.req);
|
let (req_id, parts) = (wrapped_req.id, wrapped_req.req);
|
||||||
|
|
||||||
let response = serve_all (parts).await;
|
let response = ptth::file_server::serve_all (parts).await;
|
||||||
|
|
||||||
let mut resp_req = client
|
let mut resp_req = client
|
||||||
.post (&format! ("{}/http_response/{}", SERVER_URL, req_id))
|
.post (&format! ("{}/http_response/{}", SERVER_URL, req_id))
|
||||||
|
@ -323,8 +79,8 @@ async fn main () -> Result <(), Box <dyn Error>> {
|
||||||
},
|
},
|
||||||
};
|
};
|
||||||
|
|
||||||
// Spawn another task for each request so we can immediately listen
|
// Spawn another task for each request so we can
|
||||||
// for the next connection
|
// immediately listen for the next connection
|
||||||
|
|
||||||
let client = client.clone ();
|
let client = client.clone ();
|
||||||
tokio::spawn (async move {
|
tokio::spawn (async move {
|
||||||
|
|
|
@ -0,0 +1,254 @@
|
||||||
|
// Static file server that can plug into the PTTH reverse server
|
||||||
|
|
||||||
|
use std::{
|
||||||
|
cmp::{min, max},
|
||||||
|
convert::{Infallible, TryInto},
|
||||||
|
io::SeekFrom,
|
||||||
|
path::PathBuf,
|
||||||
|
};
|
||||||
|
|
||||||
|
// file_server shouldn't depend on reqwest, but for now it
|
||||||
|
// does depend on their Body struct
|
||||||
|
use reqwest::Body;
|
||||||
|
|
||||||
|
use tokio::{
|
||||||
|
fs::{
|
||||||
|
File,
|
||||||
|
read_dir,
|
||||||
|
ReadDir,
|
||||||
|
},
|
||||||
|
io::AsyncReadExt,
|
||||||
|
sync::mpsc::{
|
||||||
|
channel,
|
||||||
|
},
|
||||||
|
};
|
||||||
|
|
||||||
|
use regex::Regex;
|
||||||
|
|
||||||
|
use crate::http_serde;
|
||||||
|
|
||||||
|
fn parse_range_header (range_str: &str) -> (Option <u64>, Option <u64>) {
|
||||||
|
use lazy_static::*;
|
||||||
|
|
||||||
|
lazy_static! {
|
||||||
|
static ref RE: Regex = Regex::new (r"^bytes=(\d*)-(\d*)$").expect ("Couldn't compile regex for Range header");
|
||||||
|
}
|
||||||
|
|
||||||
|
println! ("{}", range_str);
|
||||||
|
|
||||||
|
let caps = match RE.captures (range_str) {
|
||||||
|
Some (x) => x,
|
||||||
|
_ => return (None, None),
|
||||||
|
};
|
||||||
|
let start = caps.get (1).map (|x| x.as_str ());
|
||||||
|
let end = caps.get (2).map (|x| x.as_str ());
|
||||||
|
|
||||||
|
let start = start.map (|x| u64::from_str_radix (x, 10).ok ()).flatten ();
|
||||||
|
let end = end.map (|x| u64::from_str_radix (x, 10).ok ()).flatten ();
|
||||||
|
|
||||||
|
(start, end)
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn serve_dir (mut dir: ReadDir) -> http_serde::Response {
|
||||||
|
let (tx, rx) = channel (2);
|
||||||
|
|
||||||
|
tokio::spawn (async move {
|
||||||
|
let mut tx = tx;
|
||||||
|
|
||||||
|
tx.send (Ok::<_, Infallible> (String::from ("<style>body {font-family:sans;}</style><ul>").into_bytes ())).await.unwrap ();
|
||||||
|
|
||||||
|
while let Ok (entry) = dir.next_entry ().await {
|
||||||
|
let entry: tokio::fs::DirEntry = match entry {
|
||||||
|
Some (x) => x,
|
||||||
|
None => break,
|
||||||
|
};
|
||||||
|
|
||||||
|
let trailing_slash = match entry.file_type ().await {
|
||||||
|
Ok (t) => if t.is_dir () {
|
||||||
|
"/"
|
||||||
|
}
|
||||||
|
else {
|
||||||
|
""
|
||||||
|
},
|
||||||
|
Err (_) => "",
|
||||||
|
};
|
||||||
|
|
||||||
|
let file_name = match entry.file_name ().into_string () {
|
||||||
|
Ok (x) => x,
|
||||||
|
Err (_) => {
|
||||||
|
let s = format! ("<li>(file_name error)</li>\n");
|
||||||
|
if tx.send (Ok::<_, Infallible> (s.into_bytes ())).await.is_err ()
|
||||||
|
{
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
use std::borrow::Cow;
|
||||||
|
use percent_encoding::*;
|
||||||
|
|
||||||
|
let percent_file_name: Cow <str> = utf8_percent_encode (&file_name, CONTROLS).into ();
|
||||||
|
|
||||||
|
let s = format! ("<li><a href=\"{}{}\">{}{}</a></li>\n", percent_file_name, trailing_slash, file_name, trailing_slash);
|
||||||
|
if tx.send (Ok::<_, Infallible> (s.into_bytes ())).await.is_err ()
|
||||||
|
{
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
tx.send (Ok::<_, Infallible> (String::from ("</ul>").into_bytes ())).await.unwrap ();
|
||||||
|
});
|
||||||
|
|
||||||
|
let mut response = http_serde::Response::default ();
|
||||||
|
|
||||||
|
response.header ("content-type".into (), String::from ("text/html").into_bytes ());
|
||||||
|
response.body (Body::wrap_stream (rx));
|
||||||
|
|
||||||
|
response
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn serve_file (
|
||||||
|
mut f: File,
|
||||||
|
should_send_body: bool,
|
||||||
|
range_start: Option <u64>,
|
||||||
|
range_end: Option <u64>
|
||||||
|
) -> http_serde::Response {
|
||||||
|
let (tx, rx) = channel (2);
|
||||||
|
let body = if should_send_body {
|
||||||
|
Some (Body::wrap_stream (rx))
|
||||||
|
}
|
||||||
|
else {
|
||||||
|
None
|
||||||
|
};
|
||||||
|
|
||||||
|
let file_md = f.metadata ().await.unwrap ();
|
||||||
|
let file_len = file_md.len ();
|
||||||
|
|
||||||
|
let start = range_start.unwrap_or (0);
|
||||||
|
let end = range_end.unwrap_or (file_len);
|
||||||
|
|
||||||
|
let start = max (0, min (start, file_len));
|
||||||
|
let end = max (0, min (end, file_len));
|
||||||
|
|
||||||
|
f.seek (SeekFrom::Start (start)).await.unwrap ();
|
||||||
|
|
||||||
|
println! ("Serving range {}-{}", start, end);
|
||||||
|
|
||||||
|
if should_send_body {
|
||||||
|
tokio::spawn (async move {
|
||||||
|
//println! ("Opening file {:?}", path);
|
||||||
|
|
||||||
|
let mut tx = tx;
|
||||||
|
//let mut bytes_sent = 0;
|
||||||
|
let mut bytes_left = end - start;
|
||||||
|
|
||||||
|
loop {
|
||||||
|
let mut buffer = vec! [0u8; 65_536];
|
||||||
|
let bytes_read: u64 = f.read (&mut buffer).await.unwrap ().try_into ().unwrap ();
|
||||||
|
|
||||||
|
let bytes_read = min (bytes_left, bytes_read);
|
||||||
|
|
||||||
|
buffer.truncate (bytes_read.try_into ().unwrap ());
|
||||||
|
|
||||||
|
if bytes_read == 0 {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
if tx.send (Ok::<_, Infallible> (buffer)).await.is_err () {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
bytes_left -= bytes_read;
|
||||||
|
if bytes_left == 0 {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
//bytes_sent += bytes_read;
|
||||||
|
//println! ("Sent {} bytes", bytes_sent);
|
||||||
|
|
||||||
|
//delay_for (Duration::from_millis (50)).await;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
let mut response = http_serde::Response::default ();
|
||||||
|
|
||||||
|
response.header (String::from ("accept-ranges"), b"bytes".to_vec ());
|
||||||
|
|
||||||
|
if range_start.is_none () && range_end.is_none () {
|
||||||
|
response.status_code (http_serde::StatusCode::Ok);
|
||||||
|
response.header (String::from ("content-length"), end.to_string ().into_bytes ());
|
||||||
|
}
|
||||||
|
else {
|
||||||
|
response.status_code (http_serde::StatusCode::PartialContent);
|
||||||
|
response.header (String::from ("content-range"), format! ("bytes {}-{}/{}", start, end - 1, end).into_bytes ());
|
||||||
|
}
|
||||||
|
|
||||||
|
if let Some (body) = body {
|
||||||
|
response.body (body);
|
||||||
|
}
|
||||||
|
|
||||||
|
response
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn serve_error (
|
||||||
|
status_code: http_serde::StatusCode,
|
||||||
|
msg: String
|
||||||
|
)
|
||||||
|
-> http_serde::Response
|
||||||
|
{
|
||||||
|
let mut resp = http_serde::Response::default ();
|
||||||
|
resp.status_code (status_code)
|
||||||
|
.body (msg.into ());
|
||||||
|
resp
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn serve_all (
|
||||||
|
parts: http_serde::RequestParts
|
||||||
|
)
|
||||||
|
-> http_serde::Response
|
||||||
|
{
|
||||||
|
println! ("Client requested {}", parts.uri);
|
||||||
|
let mut range_start = None;
|
||||||
|
let mut range_end = None;
|
||||||
|
|
||||||
|
for (k, v) in parts.headers.iter () {
|
||||||
|
let v = std::str::from_utf8 (v).unwrap ();
|
||||||
|
//println! ("{}: {}", k, v);
|
||||||
|
|
||||||
|
if k == "range" {
|
||||||
|
let (start, end) = parse_range_header (v);
|
||||||
|
range_start = start;
|
||||||
|
range_end = end;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let should_send_body = match &parts.method {
|
||||||
|
http_serde::Method::Get => true,
|
||||||
|
_ => false,
|
||||||
|
};
|
||||||
|
|
||||||
|
use percent_encoding::*;
|
||||||
|
|
||||||
|
let encoded_path = &parts.uri [1..];
|
||||||
|
let path = percent_decode (encoded_path.as_bytes ()).decode_utf8 ().unwrap ();
|
||||||
|
|
||||||
|
let mut full_path = PathBuf::from ("/home/user");
|
||||||
|
full_path.push (&*path);
|
||||||
|
|
||||||
|
if let Ok (dir) = read_dir (&full_path).await {
|
||||||
|
serve_dir (dir).await
|
||||||
|
}
|
||||||
|
else if let Ok (file) = File::open (&full_path).await {
|
||||||
|
serve_file (
|
||||||
|
file,
|
||||||
|
should_send_body,
|
||||||
|
range_start,
|
||||||
|
range_end
|
||||||
|
).await
|
||||||
|
}
|
||||||
|
else {
|
||||||
|
serve_error (http_serde::StatusCode::NotFound, "404 Not Found".into ()).await
|
||||||
|
}
|
||||||
|
}
|
|
@ -1,2 +1,3 @@
|
||||||
|
pub mod file_server;
|
||||||
pub mod http_serde;
|
pub mod http_serde;
|
||||||
pub mod watcher;
|
pub mod watcher;
|
||||||
|
|
Loading…
Reference in New Issue