reactor
/
ptth
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
ptth/src/bin/server.rs

329 lines
7.0 KiB
Rust
Raw Normal View History

Concept is proven, though it won't work for streaming
2020-10-27 03:27:25 +00:00
use std::{
Byte ranges are properly advertised and served. This means Firefox can seek video / audio served by us. To think, you used to need Flash to do this.
2020-10-28 03:04:08 +00:00
cmp::{min, max},
:sparkle: Sending the ResponseParts back through a header. Do you feel like a hero yet?
2020-10-28 02:29:15 +00:00
collections::*,
Byte ranges are properly advertised and served. This means Firefox can seek video / audio served by us. To think, you used to need Flash to do this.
2020-10-28 03:04:08 +00:00
convert::{Infallible, TryInto},
Concept is proven, though it won't work for streaming
2020-10-27 03:27:25 +00:00
error::Error,
:recycle: Towards response headers and status code
2020-10-28 02:10:40 +00:00
io::SeekFrom,
:sparkle: You can now actually request different files
2020-10-28 01:39:02 +00:00
path::PathBuf,
Concept is proven, though it won't work for streaming
2020-10-27 03:27:25 +00:00
sync::Arc,
Sending a picture to the client. Streaming works on the server but not yet on the relay
2020-10-27 03:58:50 +00:00
time::Duration,
Concept is proven, though it won't work for streaming
2020-10-27 03:27:25 +00:00
};
Decided to go straight for the reverse HTTP server idea
2020-10-27 02:49:12 +00:00
use hyper::{
StatusCode,
};
:recycle: Towards response headers and status code
2020-10-28 02:10:40 +00:00
use lazy_static::*;
use regex::Regex;
Sending a picture to the client. Streaming works on the server but not yet on the relay
2020-10-27 03:58:50 +00:00
use reqwest::{
Body,
Client,
};
use tokio::{
:construction: Working on directory indexes
2020-10-29 13:19:14 +00:00
fs::{
File,
read_dir,
ReadDir,
},
Sending a picture to the client. Streaming works on the server but not yet on the relay
2020-10-27 03:58:50 +00:00
io::AsyncReadExt,
sync::mpsc::{
channel,
},
time::delay_for,
};
Decided to go straight for the reverse HTTP server idea
2020-10-27 02:49:12 +00:00
:sparkle: Sending the ResponseParts back through a header. Do you feel like a hero yet?
2020-10-28 02:29:15 +00:00
use ptth::http_serde;
:recycle: Package interesting request fields into MsgPack
2020-10-28 01:20:06 +00:00
:recycle: Towards response headers and status code
2020-10-28 02:10:40 +00:00
fn parse_range_header (range_str: &str) -> (Option <u64>, Option <u64>) {
lazy_static! {
Byte ranges are properly advertised and served. This means Firefox can seek video / audio served by us. To think, you used to need Flash to do this.
2020-10-28 03:04:08 +00:00
static ref RE: Regex = Regex::new (r"^bytes=(\d*)-(\d*)$").expect ("Couldn't compile regex for Range header");
:recycle: Towards response headers and status code
2020-10-28 02:10:40 +00:00
}
Byte ranges are properly advertised and served. This means Firefox can seek video / audio served by us. To think, you used to need Flash to do this.
2020-10-28 03:04:08 +00:00
println! ("{}", range_str);
:recycle: Towards response headers and status code
2020-10-28 02:10:40 +00:00
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)
}
:recycle:
2020-10-29 12:45:35 +00:00
const SERVER_URL: &str = "http://127.0.0.1:4000";
:construction: Working on directory indexes
2020-10-29 13:19:14 +00:00
struct ResponseHandle <'a> {
client: &'a Client,
req_id: &'a str,
}
impl <'a> ResponseHandle <'a> {
async fn respond (
self,
resp_parts: http_serde::ResponseParts,
body: Option <Body>
) {
let mut resp_req = self.client
.post (&format! ("{}/http_response/{}", SERVER_URL, self.req_id))
.header ("X-PTTH-2LJYXWC4", base64::encode (rmp_serde::to_vec (&resp_parts).unwrap ()));
if let Some (body) = body {
resp_req = resp_req.body (body);
}
//println! ("Step 6");
if let Err (e) = resp_req.send ().await {
println! ("Err: {:?}", e);
}
}
}
async fn serve_dir (
response_handle: ResponseHandle <'_>,
dir: ReadDir
:recycle:
2020-10-29 12:45:35 +00:00
) {
:construction: Working on directory indexes
2020-10-29 13:19:14 +00:00
let (tx, rx) = channel (2);
:recycle:
2020-10-29 12:45:35 +00:00
:construction: Working on directory indexes
2020-10-29 13:19:14 +00:00
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 ();
for i in 0u8..10 {
let s = format! ("<li><a href=\"{}\">Dir entry {}</a></li>\n", i, i);
if tx.send (Ok::<_, Infallible> (s.into_bytes ())).await.is_err ()
{
break;
}
}
tx.send (Ok::<_, Infallible> (String::from ("</ul>").into_bytes ())).await.unwrap ();
});
:recycle:
2020-10-29 12:45:35 +00:00
:construction: Working on directory indexes
2020-10-29 13:19:14 +00:00
let mut headers: HashMap <String, Vec <u8>> = Default::default ();
headers.insert ("content-type".into (), String::from ("text/html").into_bytes ());
let resp_parts = http_serde::ResponseParts {
status_code: http_serde::StatusCode::Ok,
headers,
};
response_handle.respond (resp_parts, Some (Body::wrap_stream (rx))).await;
:recycle:
2020-10-29 12:45:35 +00:00
}
:recycle: Extract serve_file fn
2020-10-29 12:17:40 +00:00
async fn serve_file (
:construction: Working on directory indexes
2020-10-29 13:19:14 +00:00
response_handle: ResponseHandle <'_>,
:recycle: Extract serve_file fn
2020-10-29 12:17:40 +00:00
mut f: File,
should_send_body: bool,
range_start: Option <u64>,
range_end: Option <u64>
) {
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 {
:recycle:
2020-10-29 12:45:35 +00:00
tokio::spawn (async move {
//println! ("Opening file {:?}", path);
:recycle: Extract serve_file fn
2020-10-29 12:17:40 +00:00
:recycle:
2020-10-29 12:45:35 +00:00
let mut tx = tx;
//let mut bytes_sent = 0;
let mut bytes_left = end - start;
:recycle: Extract serve_file fn
2020-10-29 12:17:40 +00:00
:recycle:
2020-10-29 12:45:35 +00:00
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);
:construction: Working on directory indexes
2020-10-29 13:19:14 +00:00
//delay_for (Duration::from_millis (50)).await;
:recycle: Extract serve_file fn
2020-10-29 12:17:40 +00:00
}
:recycle:
2020-10-29 12:45:35 +00:00
});
:recycle: Extract serve_file fn
2020-10-29 12:17:40 +00:00
}
let mut headers: HashMap <String, Vec <u8>> = Default::default ();
headers.insert (String::from ("accept-ranges"), b"bytes".to_vec ());
let status_code;
if range_start.is_none () && range_end.is_none () {
headers.insert (String::from ("content-length"), end.to_string ().into_bytes ());
status_code = http_serde::StatusCode::Ok;
}
else {
headers.insert (String::from ("content-range"), format! ("bytes {}-{}/{}", start, end - 1, end).into_bytes ());
status_code = http_serde::StatusCode::PartialContent;
}
let resp_parts = http_serde::ResponseParts {
status_code,
headers,
};
:construction: Working on directory indexes
2020-10-29 13:19:14 +00:00
response_handle.respond (resp_parts, body).await;
:recycle: Extract serve_file fn
2020-10-29 12:17:40 +00:00
}
:bug: Add 404 page for missing file instead of gateway timeout
2020-10-29 12:22:42 +00:00
async fn serve_error (
:construction: Working on directory indexes
2020-10-29 13:19:14 +00:00
response_handle: ResponseHandle <'_>,
status_code: http_serde::StatusCode
:bug: Add 404 page for missing file instead of gateway timeout
2020-10-29 12:22:42 +00:00
) {
let headers: HashMap <String, Vec <u8>> = Default::default ();
let resp_parts = http_serde::ResponseParts {
status_code,
headers,
};
:construction: Working on directory indexes
2020-10-29 13:19:14 +00:00
response_handle.respond (resp_parts, Some ("404 Not Found".into ())).await;
:recycle:
2020-10-29 12:45:35 +00:00
}
async fn handle_all (
:construction: Working on directory indexes
2020-10-29 13:19:14 +00:00
client: &Client,
:recycle:
2020-10-29 12:45:35 +00:00
req_resp: reqwest::Response
) -> Result <(), Box <dyn Error>> {
//println! ("Step 1");
if req_resp.status () != StatusCode::OK {
// TODO: Error handling
return Ok (());
}
println! ("Step 3");
let body = req_resp.bytes ().await?;
let parts: http_serde::RequestParts = match rmp_serde::from_read_ref (&body)
{
Ok (x) => x,
_ => return Ok (()),
};
let (req_id, uri) = (parts.id, parts.uri);
println! ("Client requested {}", 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,
};
:bug: Add 404 page for missing file instead of gateway timeout
2020-10-29 12:22:42 +00:00
//println! ("Step 6");
:recycle:
2020-10-29 12:45:35 +00:00
let mut path = PathBuf::from ("/home/user");
path.push (&uri [1..]);
:construction: Working on directory indexes
2020-10-29 13:19:14 +00:00
let response_handle = ResponseHandle {
client,
req_id: &req_id,
};
if let Ok (dir) = read_dir (&path).await {
serve_dir (response_handle, dir).await;
}
else if let Ok (file) = File::open (&path).await {
serve_file (
response_handle,
file,
:recycle:
2020-10-29 12:45:35 +00:00
should_send_body,
range_start,
range_end
:construction: Working on directory indexes
2020-10-29 13:19:14 +00:00
).await;
}
else {
serve_error (
response_handle,
http_serde::StatusCode::NotFound
).await;
:bug: Add 404 page for missing file instead of gateway timeout
2020-10-29 12:22:42 +00:00
}
:recycle:
2020-10-29 12:45:35 +00:00
Ok (())
:bug: Add 404 page for missing file instead of gateway timeout
2020-10-29 12:22:42 +00:00
}
Decided to go straight for the reverse HTTP server idea
2020-10-27 02:49:12 +00:00
#[tokio::main]
async fn main () -> Result <(), Box <dyn Error>> {
Concept is proven, though it won't work for streaming
2020-10-27 03:27:25 +00:00
let client = Arc::new (Client::new ());
Decided to go straight for the reverse HTTP server idea
2020-10-27 02:49:12 +00:00
Sending a picture to the client. Streaming works on the server but not yet on the relay
2020-10-27 03:58:50 +00:00
let mut backoff_delay = 0;
Decided to go straight for the reverse HTTP server idea
2020-10-27 02:49:12 +00:00
loop {
Sending a picture to the client. Streaming works on the server but not yet on the relay
2020-10-27 03:58:50 +00:00
if backoff_delay > 0 {
delay_for (Duration::from_millis (backoff_delay)).await;
}
Concept is proven, though it won't work for streaming
2020-10-27 03:27:25 +00:00
let req_req = client.get ("http://127.0.0.1:4000/http_listen/alien_wildlands");
let req_resp = match req_req.send ().await {
Decided to go straight for the reverse HTTP server idea
2020-10-27 02:49:12 +00:00
Err (e) => {
println! ("Err: {:?}", e);
Sending a picture to the client. Streaming works on the server but not yet on the relay
2020-10-27 03:58:50 +00:00
backoff_delay = backoff_delay * 2 + 500;
Decided to go straight for the reverse HTTP server idea
2020-10-27 02:49:12 +00:00
continue;
},
Sending a picture to the client. Streaming works on the server but not yet on the relay
2020-10-27 03:58:50 +00:00
Ok (r) => {
backoff_delay = 0;
r
},
Decided to go straight for the reverse HTTP server idea
2020-10-27 02:49:12 +00:00
};
:recycle:
2020-10-29 12:45:35 +00:00
// Spawn another task for each request so we can immediately listen
// for the next connection
:sparkle: You can now actually request different files
2020-10-28 01:39:02 +00:00
Concept is proven, though it won't work for streaming
2020-10-27 03:27:25 +00:00
let client = client.clone ();
tokio::spawn (async move {
:construction: Working on directory indexes
2020-10-29 13:19:14 +00:00
match handle_all (&client, req_resp).await {
_ => (),
}
Concept is proven, though it won't work for streaming
2020-10-27 03:27:25 +00:00
});
Decided to go straight for the reverse HTTP server idea
2020-10-27 02:49:12 +00:00
}
}