Support multi threaded partial-content download.

Author: agrawal-d

2-mb-file.txt

@@ -9,6 +9,8 @@ pub struct HTTPMessage {
 
 impl HTTPMessage {
     pub fn new(raw_message: &str) -> HTTPMessage {
+        println!("Raw message while parsing \n`{}`\n", raw_message);
+
         let (headers, body) = raw_message.split_once("\r\n\r\n").unwrap();
         let lines = headers.lines().enumerate();
         let mut headers: HashMap<String, String> = HashMap::new();

Cargo.lock

@@ -1,6 +1,12 @@
 mod http_message;
 pub mod tcp;
+use std::fs;
+use std::{cmp::min, str, thread, time};
+
 pub use crate::http_message::HTTPMessage;
+use crate::tcp::{download_part, head};
+
+static DIVISIONS: usize = 302400;
 
 pub fn find_subsequence<T>(haystack: &[T], needle: &[T]) -> Option<usize>
 where
@@ -10,3 +16,57 @@ where
         .windows(needle.len())
         .position(|window| window == needle)
 }
+
+pub fn download(server_details: &str, path: &str) {
+    let head_req = head(&server_details, path);
+    // println!("{:?}", head_req);
+
+    if let Some(value) = head_req.headers.get("Accept-Ranges") {
+        if value != "bytes" {
+            eprintln!("Server does not support downloading byte ranges");
+            return;
+        }
+
+        let content_length: usize =
+            str::parse(head_req.headers.get("Content-Length").unwrap()).unwrap();
+
+        println!("Content-Length: {}", content_length);
+
+        let mut buf: Vec<u8> = Vec::with_capacity(content_length);
+        let mut threads = Vec::new();
+
+        let mut consumed = 0;
+        let mut i = 0;
+        while consumed < content_length {
+            println!("{}", i);
+            i += 1;
+            let address = server_details.to_string();
+            let path = path.to_string();
+
+            threads.push(thread::spawn(move || {
+                let data = download_part(
+                    address,
+                    path,
+                    consumed,
+                    min(consumed + DIVISIONS, content_length - 1),
+                );
+                data
+            }));
+
+            // let one_second = time::Duration::from_secs(1);
+            // thread::sleep(one_second);
+
+            consumed += DIVISIONS;
+        }
+
+        for handle in threads {
+            let data = handle.join().unwrap();
+            buf.extend(data);
+        }
+
+        fs::write("data.txt", buf).unwrap();
+    } else {
+        eprintln!("Mult-threaded downloads not supported for this request.");
+        return;
+    }
+}

src/http_message.rs

@@ -1,16 +1,12 @@
-use multidl::{find_subsequence, tcp::download_part};
-
-use std::net::{TcpListener, TcpStream, ToSocketAddrs};
+use multidl::download;
+use std::env;
 
 fn main() {
-    let server_details = "example.com:80";
-    let server: Vec<_> = server_details
-        .to_socket_addrs()
-        .expect("Unable to resolve domain")
-        .collect();
-    println!("{:?}", server);
+    let args: Vec<String> = env::args().collect();
+    println!("{:?}", args);
+
+    let server_details = &args[1];
+    let path = &args[2];
 
-    // Even easier, if you want to connect right away:
-    let stream = TcpStream::connect(server_details).expect("Unable to connect to server");
-    // let listener = TcpListener::bind(server[1]).unwrap();
+    download(server_details, path)
 }

src/lib.rs

@@ -1,51 +1,71 @@
 use std::{
-    fmt::format,
     io::{Read, Write},
-    net::TcpListener,
+    net::TcpStream,
+    str,
 };
 
-use std::net::{SocketAddr, ToSocketAddrs};
+use crate::{find_subsequence, HTTPMessage};
 
-use std::str;
+static BUFFER_SIZE: usize = 409600;
 
-use crate::{find_subsequence, HTTPMessage};
+pub fn head(address: &str, path: &str) -> HTTPMessage {
+    let mut stream = TcpStream::connect(address).expect("Unable to connect to server");
+    let (host, _) = address.split_once(":").unwrap();
+    let request = format!(
+        "HEAD {} HTTP/1.1\r\nHost: {}\r\nAccept: */*\r\n\r\n",
+        path, host
+    );
+    stream.write(request.as_bytes()).unwrap();
+    let mut buffer: Vec<u8> = vec![0; 1024];
+    stream.read(&mut buffer).unwrap();
 
-pub fn download_part(
-    address: &str,
-    path: &str,
-    bytes_from: usize,
-    bytes_to: usize,
-) -> Option<Vec<u8>> {
-    let addr = address.to_socket_addrs().unwrap().next().unwrap();
-    println!("{:?}", addr);
-    let listener = TcpListener::bind(addr).unwrap();
+    HTTPMessage::new(str::from_utf8(&buffer).unwrap())
+}
+
+pub fn download_part(address: String, path: String, bytes_from: usize, bytes_to: usize) -> Vec<u8> {
+    let mut stream = TcpStream::connect(&address).expect("Unable to connect to server");
     let (host, _) = address.split_once(":").unwrap();
-    for stream in listener.incoming() {
-        match stream {
-            Ok(mut stream) => {
-                let request = format!(
-                    "GET {} HTTP/1.1\r\nHost: {}\r\nRange: bytes={}-{}\r\nAccept: */*\r\n\r\n",
-                    path, host, bytes_from, bytes_to
-                );
-                stream.write(request.as_bytes());
-
-                let body_size = bytes_to - bytes_from + 1;
-                let mut buffer: Vec<u8> = Vec::with_capacity(body_size + 1024);
-                let body_break: Vec<u8> = vec![13, 10];
-                stream.read_to_end(&mut buffer).unwrap();
-
-                let loc_body_break = find_subsequence(&buffer, &body_break).unwrap();
-                let metadata = str::from_utf8(&buffer[0..loc_body_break + 2]).unwrap();
-                let message = HTTPMessage::new(metadata);
-                println!("{:?}", message);
-            }
-            Err(e) => {
-                eprintln!("Failed to get stream from Tcp Listener : {}", e);
-            }
+    let request = format!(
+        "GET {} HTTP/1.1\r\nHost: {}\r\nRange: bytes={}-{}\r\nAccept: */*\r\n\r\n",
+        path, host, bytes_from, bytes_to
+    );
+    stream.write(request.as_bytes()).unwrap();
+
+    let body_size = bytes_to - bytes_from + 1;
+    let mut buffer: Vec<u8> = vec![0; BUFFER_SIZE];
+    let mut body: Vec<u8> = Vec::with_capacity(body_size);
+    let body_break: Vec<u8> = "\r\n\r\n".as_bytes().to_vec();
+    let mut content_consumed: usize = 0;
+
+    let mut first = true;
+
+    while content_consumed < body_size {
+        let bytes_read = stream.read(&mut buffer).unwrap();
+        println!("Read {} bytes", bytes_read);
+
+        if bytes_read == 0 {
+            panic!("WTF");
         }
 
-        return None;
+        if first {
+            first = false;
+            let loc_body_break = find_subsequence(&buffer, &body_break).unwrap();
+            let metadata = str::from_utf8(&buffer[0..loc_body_break + 4]).unwrap();
+            let message = HTTPMessage::new(metadata);
+            let _content_length: usize =
+                str::parse(message.headers.get("Content-Length").unwrap()).unwrap();
+            let content = &buffer[loc_body_break + 4..bytes_read];
+            content_consumed += content.len();
+            body.extend_from_slice(content);
+        } else {
+            content_consumed += bytes_read;
+            body.extend_from_slice(&buffer[0..bytes_read]);
+        }
+
+        println!("Consumed {} bytes of total {}", content_consumed, body_size);
     }
 
-    unimplemented!()
+    assert_eq!(body.len(), body_size);
+
+    return body;
 }

src/main.rs

@@ -1 +1,6 @@
-connect
+        // println!(
+        //     "{}/{} -> `{}`",
+        //     num_bytes_read_now,
+        //     body_size,
+        //     str::from_utf8(&buffer).unwrap()
+        // );