♻️ Towards response headers and status code

main
_ 2020-10-27 21:10:40 -05:00
parent d286adadc9
commit 1e86e9735e
5 changed files with 62 additions and 16 deletions

View File

@ -9,9 +9,12 @@ edition = "2018"
[dependencies] [dependencies]
base64 = "0.12"
futures = "0.3" futures = "0.3"
http = "0.2" http = "0.2"
hyper = "0.13" hyper = "0.13"
lazy_static = "1.4"
regex = "1"
reqwest = { version = "0.10.8", features = ["stream"] } reqwest = { version = "0.10.8", features = ["stream"] }
rmp-serde = "0.14.4" rmp-serde = "0.14.4"
serde = {version = "1.0", features = ["derive"]} serde = {version = "1.0", features = ["derive"]}

View File

@ -173,7 +173,12 @@ async fn handle_http_request (
match r.await { match r.await {
Ok (Message::HttpResponseResponseStream (body)) => { Ok (Message::HttpResponseResponseStream (body)) => {
println! ("Step 7"); println! ("Step 7");
status_reply (StatusCode::OK, body)
Response::builder ()
.status (StatusCode::OK)
.header ("Accept-Ranges", "bytes")
.body (body)
.unwrap ()
}, },
_ => status_reply (StatusCode::GATEWAY_TIMEOUT, "server didn't reply in time or somethin'"), _ => status_reply (StatusCode::GATEWAY_TIMEOUT, "server didn't reply in time or somethin'"),
} }

View File

@ -1,6 +1,7 @@
use std::{ use std::{
convert::Infallible, convert::Infallible,
error::Error, error::Error,
io::SeekFrom,
path::PathBuf, path::PathBuf,
sync::Arc, sync::Arc,
time::Duration, time::Duration,
@ -10,6 +11,8 @@ use hyper::{
StatusCode, StatusCode,
Uri, Uri,
}; };
use lazy_static::*;
use regex::Regex;
use reqwest::{ use reqwest::{
Body, Body,
Client, Client,
@ -25,6 +28,24 @@ use tokio::{
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"^(\d+)-(\d+)$").expect ("Couldn't compile regex for Range header");
}
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)
}
#[tokio::main] #[tokio::main]
async fn main () -> Result <(), Box <dyn Error>> { async fn main () -> Result <(), Box <dyn Error>> {
let client = Arc::new (Client::new ()); let client = Arc::new (Client::new ());
@ -73,6 +94,19 @@ async fn main () -> Result <(), Box <dyn Error>> {
let (req_id, uri) = (parts.id, parts.uri); let (req_id, uri) = (parts.id, parts.uri);
println! ("Client requested {}", 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;
}
}
println! ("Step 4/5"); println! ("Step 4/5");
@ -80,22 +114,23 @@ async fn main () -> Result <(), Box <dyn Error>> {
let client = client.clone (); let client = client.clone ();
tokio::spawn (async move { tokio::spawn (async move {
let (tx, rx) = channel (2); let (tx, rx) = channel (2);
//let rx: Receiver <Vec <u8>> = rx;
tokio::spawn (async move {
//let path = "/home/user/projects/2020/ptth/README.md";
let mut path = PathBuf::from ("/home/user"); let mut path = PathBuf::from ("/home/user");
path.push (&uri [1..]); path.push (&uri [1..]);
println! ("Opening file {:?}", path);
let mut f = File::open (path).await.unwrap (); let mut f = File::open (path).await.unwrap ();
if let Some (start) = range_start {
f.seek (SeekFrom::Start (start)).await.unwrap ();
}
tokio::spawn (async move {
//println! ("Opening file {:?}", path);
let mut tx = tx; let mut tx = tx;
let mut bytes_sent = 0; //let mut bytes_sent = 0;
loop { loop {
let mut buffer = vec! [0u8; 4096]; let mut buffer = vec! [0u8; 65_536];
let bytes_read = f.read (&mut buffer).await.unwrap (); let bytes_read = f.read (&mut buffer).await.unwrap ();
buffer.truncate (bytes_read); buffer.truncate (bytes_read);
@ -104,10 +139,12 @@ async fn main () -> Result <(), Box <dyn Error>> {
break; break;
} }
tx.send (Ok::<_, Infallible> (buffer)).await.unwrap (); if tx.send (Ok::<_, Infallible> (buffer)).await.is_err () {
bytes_sent += bytes_read; break;
}
println! ("Sent {} bytes", bytes_sent); //bytes_sent += bytes_read;
//println! ("Sent {} bytes", bytes_sent);
delay_for (Duration::from_millis (50)).await; delay_for (Duration::from_millis (50)).await;
} }

View File

@ -62,7 +62,7 @@ pub enum StatusCode {
Ok, Ok,
NotFound, NotFound,
} }
/*
impl TryFrom <hyper::StatusCode> for StatusCode { impl TryFrom <hyper::StatusCode> for StatusCode {
type Error = Error; type Error = Error;
@ -74,7 +74,7 @@ impl TryFrom <hyper::StatusCode> for StatusCode {
} }
} }
} }
*/
impl From <StatusCode> for hyper::StatusCode { impl From <StatusCode> for hyper::StatusCode {
fn from (x: StatusCode) -> Self { fn from (x: StatusCode) -> Self {
match x { match x {

View File

@ -1,2 +1,3 @@
- Figure out how to pack response status / headers into request body
- Byte range request header - Byte range request header
- Content-Length response header - Content-Length response header