pub async fn register(&mut self, file_description: Box<dyn FileDescription>) -> FileManager {
        let register_time = Instant::now();
        let (tx, mut rx) = unbounded_channel::<Range>();
        let file_cache;
        {
            let mut data_guard = self.data.lock().unwrap();
            file_cache = data_guard
                .entry(CacheKey {
                    downloader_id: file_description.get_downloader_id(),
                    uri: file_description.get_uri(),
                })
                .or_insert_with(|| FileCache::new(file_description.get_file_size()))
                .clone();
        }
        let file_manager = FileManager::new(file_cache.clone(), tx);
        let downloader_ref = self.register_downloader(&*file_description);
        let semaphore_ref = Arc::clone(&self.semaphore);
        let stat_ref = self.stats.clone();
        let release_rate = self.release_rate;
        let absolute_max_parallel = self.absolute_max_parallel;
        let current_max_parallel = Arc::clone(&self.current_max_parallel);
        let uri = file_description.get_uri();
        tokio::spawn(async move {
            while let Some(message) = rx.recv().await {
                // obtain a permit, it will be released once the download completes
                let permit = semaphore_ref.acquire().await.unwrap();
                permit.forget();
                // run download in a dedicated task
                let downloader_ref = Arc::clone(&downloader_ref);
                let semaphore_ref = Arc::clone(&semaphore_ref);
                let current_max_parallel = Arc::clone(&current_max_parallel);
                let stats_ref = stat_ref.clone();
                let file_cache = file_cache.clone();
                let uri = uri.clone();
                tokio::spawn(async move {
                    // start the actual download
                    let dl_start_time = Instant::now();

                    let dl_res = downloader_ref
                        .download(uri.clone(), message.start, message.length)
                        .await;
                    // once the download is completed, release the permit at the configured rate
                    let new_permits = DownloadCache::permit_leap(
                        &current_max_parallel,
                        absolute_max_parallel,
                        release_rate,
                    );
                    semaphore_ref.add_permits(new_permits);
                    let dl_enum = match dl_res {
                        Ok(downloaded_chunk) => {
                            stats_ref.record_download(DownloadStat {
                                dl_duration: dl_start_time.elapsed().as_millis() as u64,
                                size: downloaded_chunk.len() as u64,
                                dl_start: dl_start_time.duration_since(register_time).as_millis()
                                    as u64,
                            });
                            Download::Done(Arc::new(downloaded_chunk))
                        }
                        Err(err) => Download::Error(format!("{:?}", err)),
                    };
                    file_cache.insert(message.start, dl_enum);
                });
            }
        });
        file_manager
    }
/// Converts a [`FileDescription`] trait object into a [`FileManager`].
/// Spawns a task that will download the file chunks queued on the [`FileManager`].
/// TODO do not re-download chunks if same file was already registered
Converts a [`FileDescription`] trait object into a [`FileManager`].
Spawns a task that will download the file chunks queued on the [`FileManager`].
TODO do not re-download chunks if same file was already registered
Converts a [ ` FileDescription ` ] trait object into a [ ` FileManager ` ] . Spawns a task that will download the file chunks queued on the [ ` FileManager ` ] . TODO do not re - download chunks if same file was already registered
pub async fn register(&mut self, file_description: Box<dyn FileDescription>) -> FileManager {
        let register_time = Instant::now();
        let (tx, mut rx) = unbounded_channel::<Range>();
        let file_cache;
        {
            let mut data_guard = self.data.lock().unwrap();
            file_cache = data_guard
                .entry(CacheKey {
                    downloader_id: file_description.get_downloader_id(),
                    uri: file_description.get_uri(),
                })
                .or_insert_with(|| FileCache::new(file_description.get_file_size()))
                .clone();
        }
        let file_manager = FileManager::new(file_cache.clone(), tx);
        let downloader_ref = self.register_downloader(&*file_description);
        let semaphore_ref = Arc::clone(&self.semaphore);
        let stat_ref = self.stats.clone();
        let release_rate = self.release_rate;
        let absolute_max_parallel = self.absolute_max_parallel;
        let current_max_parallel = Arc::clone(&self.current_max_parallel);
        let uri = file_description.get_uri();
        tokio::spawn(async move {
            while let Some(message) = rx.recv().await {
                // obtain a permit, it will be released once the download completes
                let permit = semaphore_ref.acquire().await.unwrap();
                permit.forget();
                // run download in a dedicated task
                let downloader_ref = Arc::clone(&downloader_ref);
                let semaphore_ref = Arc::clone(&semaphore_ref);
                let current_max_parallel = Arc::clone(&current_max_parallel);
                let stats_ref = stat_ref.clone();
                let file_cache = file_cache.clone();
                let uri = uri.clone();
                tokio::spawn(async move {
                    // start the actual download
                    let dl_start_time = Instant::now();
                    let dl_res = downloader_ref
                        .download(uri.clone(), message.start, message.length)
                        .await;
                    // once the download is completed, release the permit at the configured rate
                    let new_permits = DownloadCache::permit_leap(
                        &current_max_parallel,
                        absolute_max_parallel,
                        release_rate,
                    );
                    semaphore_ref.add_permits(new_permits);
                    let dl_enum = match dl_res {
                        Ok(downloaded_chunk) => {
                            stats_ref.record_download(DownloadStat {
                                dl_duration: dl_start_time.elapsed().as_millis() as u64,
                                size: downloaded_chunk.len() as u64,
                                dl_start: dl_start_time.duration_since(register_time).as_millis()
                                    as u64,
                            });
                            Download::Done(Arc::new(downloaded_chunk))
                        }
                        Err(err) => Download::Error(format!("{:?}", err)),
                    };
                    file_cache.insert(message.start, dl_enum);
                });
            }
        });
        file_manager
    }
pub async fn register ( & mut self , file_description : Box < dyn FileDescription > ) -> FileManager { let register_time = Instant :: now ( ) ; let ( tx , mut rx ) = unbounded_channel :: < Range > ( ) ; let file_cache ; { let mut data_guard = self . data . lock ( ) . unwrap ( ) ; file_cache = data_guard . entry ( CacheKey { downloader_id : file_description . get_downloader_id ( ) , uri : file_description . get_uri ( ) , } ) . or_insert_with ( | | FileCache :: new ( file_description . get_file_size ( ) ) ) . clone ( ) ; } let file_manager = FileManager :: new ( file_cache . clone ( ) , tx ) ; let downloader_ref = self . register_downloader ( & * file_description ) ; let semaphore_ref = Arc :: clone ( & self . semaphore ) ; let stat_ref = self . stats . clone ( ) ; let release_rate = self . release_rate ; let absolute_max_parallel = self . absolute_max_parallel ; let current_max_parallel = Arc :: clone ( & self . current_max_parallel ) ; let uri = file_description . get_uri ( ) ; tokio :: spawn ( async move { while let Some ( message ) = rx . recv ( ) . await { // obtain a permit, it will be released once the download completes let permit = semaphore_ref . acquire ( ) . await . unwrap ( ) ; permit . forget ( ) ; // run download in a dedicated task let downloader_ref = Arc :: clone ( & downloader_ref ) ; let semaphore_ref = Arc :: clone ( & semaphore_ref ) ; let current_max_parallel = Arc :: clone ( & current_max_parallel ) ; let stats_ref = stat_ref . clone ( ) ; let file_cache = file_cache . clone ( ) ; let uri = uri . clone ( ) ; tokio :: spawn ( async move { // start the actual download let dl_start_time = Instant :: now ( ) ; let dl_res = downloader_ref . download ( uri . clone ( ) , message . start , message . length ) . await ; // once the download is completed, release the permit at the configured rate let new_permits = DownloadCache :: permit_leap ( & current_max_parallel , absolute_max_parallel , release_rate , ) ; semaphore_ref . add_permits ( new_permits ) ; let dl_enum = match dl_res { Ok ( downloaded_chunk ) => { stats_ref . record_download ( DownloadStat { dl_duration : dl_start_time . elapsed ( ) . as_millis ( ) as u64 , size : downloaded_chunk . len ( ) as u64 , dl_start : dl_start_time . duration_since ( register_time ) . as_millis ( ) as u64 , } ) ; Download :: Done ( Arc :: new ( downloaded_chunk ) ) } Err ( err ) => Download :: Error ( format ! ( " " err ) ) , } ; file_cache . insert ( message . start , dl_enum ) ; } ) ; } } ) ; file_manager }
Converts a [`FileDescription`] trait object into a [`FileManager`].
Converts a [ ` FileDescription ` ] trait object into a [ ` FileManager ` ] .
&mut self
file_description
Box<dyn FileDescription>
&mut self
file_description
Box<dyn FileDescription>