async fn download_media_concurrent<P: AsRef<Path>>(
        &self,
        media: &Media,
        path: P,
        workers: usize,
    ) -> Result<(), io::Error> {
        let document = match media {
            Media::Document(document) => document,
            _ => panic!("Only Document type is supported!"),
        };
        let size = document.size();
        let location = media.to_input_location().unwrap();
        // Allocate
        let mut file = fs::File::create(path).await?;
        file.set_len(size as u64).await?;
        file.seek(SeekFrom::Start(0)).await?;

        // Start workers
        let (tx, mut rx) = unbounded_channel();
        let part_index = Arc::new(tokio::sync::Mutex::new(0));
        let mut tasks = vec![];
        for _ in 0..workers {
            let location = location.clone();
            let tx = tx.clone();
            let part_index = part_index.clone();
            let client = self.clone();
            let task = tokio::task::spawn(async move {
                let mut retry_offset = None;
                let mut retry_counter = 0;
                loop {
                    // Calculate file offset
                    let offset = {
                        if let Some(offset) = retry_offset {
                            retry_offset = None;
                            offset
                        } else {
                            retry_counter = 0;
                            let mut i = part_index.lock().await;
                            *i += 1;
                            MAX_CHUNK_SIZE * (*i - 1)
                        }
                    };
                    if offset > size {
                        break;
                    }
                    // Fetch from telegram
                    let res = client
                        .invoke(&tl::functions::upload::GetFile {
                            precise: true,
                            cdn_supported: false,
                            location: location.clone(),
                            offset,
                            limit: MAX_CHUNK_SIZE,
                        })
                        .await;
                    match res {
                        Ok(tl::enums::upload::File::File(file)) => {
                            tx.send((offset as u64, file.bytes)).unwrap();
                        }
                        Ok(tl::enums::upload::File::CdnRedirect(_)) => {
                            panic!(
                                "API returned File::CdnRedirect even though cdn_supported = false"
                            );
                        }
                        Err(InvocationError::Rpc(err)) => {
                            // Retry on rate limit
                            if err.code == 420 {
                                tokio::time::sleep(RATE_LIMIT_DELAY).await;
                                retry_offset = Some(offset);
                                retry_counter += 1;
                                if retry_counter <= RATE_LIMIT_RETRIES {
                                    continue;
                                }
                            }
                            return Err(InvocationError::Rpc(err));
                        }
                        Err(e) => return Err(e),
                    }
                }
                Ok::<(), InvocationError>(())
            });
            tasks.push(task);
        }
        drop(tx);

        // File write loop
        let mut pos = 0;
        while let Some((offset, data)) = rx.recv().await {
            if offset != pos {
                file.seek(SeekFrom::Start(offset)).await?;
            }
            file.write_all(&data).await?;
            pos = offset + data.len() as u64;
        }

        // Check if all tasks finished succesfully
        for task in tasks {
            task.await?
                .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
        }
        Ok(())
    }
/// Downloads a `Document` to specified path using multiple connections
Downloads a `Document` to specified path using multiple connections
Downloads a ` Document ` to specified path using multiple connections
async fn download_media_concurrent<P: AsRef<Path>>(
        &self,
        media: &Media,
        path: P,
        workers: usize,
    ) -> Result<(), io::Error> {
        let document = match media {
            Media::Document(document) => document,
            _ => panic!("Only Document type is supported!"),
        };
        let size = document.size();
        let location = media.to_input_location().unwrap();
        // Allocate
        let mut file = fs::File::create(path).await?;
        file.set_len(size as u64).await?;
        file.seek(SeekFrom::Start(0)).await?;
        // Start workers
        let (tx, mut rx) = unbounded_channel();
        let part_index = Arc::new(tokio::sync::Mutex::new(0));
        let mut tasks = vec![];
        for _ in 0..workers {
            let location = location.clone();
            let tx = tx.clone();
            let part_index = part_index.clone();
            let client = self.clone();
            let task = tokio::task::spawn(async move {
                let mut retry_offset = None;
                let mut retry_counter = 0;
                loop {
                    // Calculate file offset
                    let offset = {
                        if let Some(offset) = retry_offset {
                            retry_offset = None;
                            offset
                        } else {
                            retry_counter = 0;
                            let mut i = part_index.lock().await;
                            *i += 1;
                            MAX_CHUNK_SIZE * (*i - 1)
                        }
                    };
                    if offset > size {
                        break;
                    }
                    // Fetch from telegram
                    let res = client
                        .invoke(&tl::functions::upload::GetFile {
                            precise: true,
                            cdn_supported: false,
                            location: location.clone(),
                            offset,
                            limit: MAX_CHUNK_SIZE,
                        })
                        .await;
                    match res {
                        Ok(tl::enums::upload::File::File(file)) => {
                            tx.send((offset as u64, file.bytes)).unwrap();
                        }
                        Ok(tl::enums::upload::File::CdnRedirect(_)) => {
                            panic!(
                                "API returned File::CdnRedirect even though cdn_supported = false"
                            );
                        }
                        Err(InvocationError::Rpc(err)) => {
                            // Retry on rate limit
                            if err.code == 420 {
                                tokio::time::sleep(RATE_LIMIT_DELAY).await;
                                retry_offset = Some(offset);
                                retry_counter += 1;
                                if retry_counter <= RATE_LIMIT_RETRIES {
                                    continue;
                                }
                            }
                            return Err(InvocationError::Rpc(err));
                        }
                        Err(e) => return Err(e),
                    }
                }
                Ok::<(), InvocationError>(())
            });
            tasks.push(task);
        }
        drop(tx);
        // File write loop
        let mut pos = 0;
        while let Some((offset, data)) = rx.recv().await {
            if offset != pos {
                file.seek(SeekFrom::Start(offset)).await?;
            }
            file.write_all(&data).await?;
            pos = offset + data.len() as u64;
        }
        // Check if all tasks finished succesfully
        for task in tasks {
            task.await?
                .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
        }
        Ok(())
    }
async fn download_media_concurrent < P : AsRef < Path > > ( & self , media : & Media , path : P , workers : usize , ) -> Result < ( ) , io :: Error > { let document = match media { Media :: Document ( document ) => document , _ => panic ! ( " " ) , } ; let size = document . size ( ) ; let location = media . to_input_location ( ) . unwrap ( ) ; // Allocate let mut file = fs :: File :: create ( path ) . await ? ; file . set_len ( size as u64 ) . await ? ; file . seek ( SeekFrom :: Start ( 0 ) ) . await ? ; // Start workers let ( tx , mut rx ) = unbounded_channel ( ) ; let part_index = Arc :: new ( tokio :: sync :: Mutex :: new ( 0 ) ) ; let mut tasks = vec ! [ ] ; for _ in 0 .. workers { let location = location . clone ( ) ; let tx = tx . clone ( ) ; let part_index = part_index . clone ( ) ; let client = self . clone ( ) ; let task = tokio :: task :: spawn ( async move { let mut retry_offset = None ; let mut retry_counter = 0 ; loop { // Calculate file offset let offset = { if let Some ( offset ) = retry_offset { retry_offset = None ; offset } else { retry_counter = 0 ; let mut i = part_index . lock ( ) . await ; * i += 1 ; MAX_CHUNK_SIZE * ( * i - 1 ) } } ; if offset > size { break ; } // Fetch from telegram let res = client . invoke ( & tl :: functions :: upload :: GetFile { precise : true , cdn_supported : false , location : location . clone ( ) , offset , limit : MAX_CHUNK_SIZE , } ) . await ; match res { Ok ( tl :: enums :: upload :: File :: File ( file ) ) => { tx . send ( ( offset as u64 , file . bytes ) ) . unwrap ( ) ; } Ok ( tl :: enums :: upload :: File :: CdnRedirect ( _ ) ) => { panic ! ( " " ) ; } Err ( InvocationError :: Rpc ( err ) ) => { // Retry on rate limit if err . code == 420 { tokio :: time :: sleep ( RATE_LIMIT_DELAY ) . await ; retry_offset = Some ( offset ) ; retry_counter += 1 ; if retry_counter <= RATE_LIMIT_RETRIES { continue ; } } return Err ( InvocationError :: Rpc ( err ) ) ; } Err ( e ) => return Err ( e ) , } } Ok :: < ( ) , InvocationError > ( ( ) ) } ) ; tasks . push ( task ) ; } drop ( tx ) ; // File write loop let mut pos = 0 ; while let Some ( ( offset , data ) ) = rx . recv ( ) . await { if offset != pos { file . seek ( SeekFrom :: Start ( offset ) ) . await ? ; } file . write_all ( & data ) . await ? ; pos = offset + data . len ( ) as u64 ; } // Check if all tasks finished succesfully for task in tasks { task . await ? . map_err ( | e | io :: Error :: new ( io :: ErrorKind :: Other , e ) ) ? ; } Ok ( ( ) ) }
Downloads a `Document` to specified path using multiple connections
Downloads a ` Document ` to specified path using multiple connections
&self
media
&Media
path
P
workers
usize
&self
media
&Media
path
P
workers
usize