feat: intercept http request to download task by p2p in proxy (dragon…

…flyoss#234)

Signed-off-by: Gaius <gaius.qi@gmail.com>

Cargo.toml

@@ -85,3 +85,4 @@ hyper-rustls = "0.26"
 http-body-util = "0.1.0"
 regex = "1.10.2"
 http-range-header = "0.4.0"
+futures-util = "0.3.30"

src/bin/dfdaemon/main.rs

@@ -195,6 +195,7 @@ async fn main() -> Result<(), anyhow::Error> {
     // Initialize proxy server.
     let proxy = Proxy::new(
         config.clone(),
+        task.clone(),
         shutdown.clone(),
         shutdown_complete_tx.clone(),
     );

src/grpc/dfdaemon_download.rs

@@ -16,7 +16,7 @@
 
 use crate::shutdown;
 use crate::task;
-use crate::utils::http::hashmap_to_headermap;
+use crate::utils::http::{get_range, hashmap_to_headermap};
 use crate::Result as ClientResult;
 use dragonfly_api::common::v2::Task;
 use dragonfly_api::dfdaemon::v2::{
@@ -150,7 +150,7 @@ impl DfdaemonDownload for DfdaemonDownloadServerHandler {
         let request = request.into_inner();
 
         // Check whether the download is empty.
-        let download = request
+        let mut download = request
             .download
             .ok_or(Status::invalid_argument("missing download"))?;
 
@@ -204,6 +204,20 @@ impl DfdaemonDownload for DfdaemonDownloadServerHandler {
                 Status::internal(e.to_string())
             })?;
 
+        // Download's range priority is higher than the request header's range.
+        // If download protocol is http, use the range of the request header.
+        // If download protocol is not http, use the range of the download.
+        if download.range.is_none() {
+            let content_length = task
+                .content_length()
+                .ok_or(Status::internal("missing content length in the response"))?;
+
+            download.range = get_range(&request_header, content_length).map_err(|err| {
+                error!("get range failed: {}", err);
+                Status::failed_precondition(err.to_string())
+            })?;
+        }
+
         // Clone the task.
         let task_manager = self.task.clone();
 

src/grpc/dfdaemon_upload.rs

@@ -277,7 +277,12 @@ impl DfdaemonUpload for DfdaemonUploadServerHandler {
         let mut reader = self
             .task
             .piece
-            .upload_from_local_peer_into_async_read(task_id.as_str(), piece_number, piece.length)
+            .upload_from_local_peer_into_async_read(
+                task_id.as_str(),
+                piece_number,
+                piece.length,
+                false,
+            )
             .await
             .map_err(|err| {
                 error!("read piece content from local storage: {}", err);

src/proxy/header.rs

@@ -15,9 +15,7 @@
  */
 
 use crate::config;
-use crate::utils::http::parse_range_header;
-use crate::Result;
-use dragonfly_api::common::v2::{Priority, Range};
+use dragonfly_api::common::v2::Priority;
 use reqwest::header::HeaderMap;
 use tracing::error;
 
@@ -43,17 +41,6 @@ pub const DRAGONFLY_FILTERED_QUERY_PARAMS_HEADER: &str = "X-Dragonfly-Filtered-Q
 // it specifies the piece length of the task.
 pub const DRAGONFLY_PIECE_LENGTH_HEADER: &str = "X-Dragonfly-Piece-Length";
 
-// get_range gets the range from http header.
-pub fn get_range(header: &HeaderMap, content_length: u64) -> Result<Option<Range>> {
-    match header.get(reqwest::header::RANGE) {
-        Some(range) => {
-            let range = range.to_str()?;
-            Ok(Some(parse_range_header(range, content_length)?))
-        }
-        None => Ok(None),
-    }
-}
-
 // get_tag gets the tag from http header.
 pub fn get_tag(header: &HeaderMap) -> Option<String> {
     match header.get(DRAGONFLY_TAG_HEADER) {